Web Scraper with Retry and Circuit Breaker

Problem Statement

Build a robust asynchronous web scraper that fetches data from multiple URLs concurrently with sophisticated error handling. The scraper must handle transient failures (timeouts, network errors) with exponential backoff retry, prevent cascading failures with circuit breakers, and aggregate results from parallel operations.

Your scraper should support:

• Fetching multiple URLs concurrently (using tokio or async-std)
• Retrying failed requests with exponential backoff (up to N attempts)
• Implementing circuit breaker pattern (open/half-open/closed states)
• Handling timeouts on all network operations
• Collecting partial results (some URLs may fail permanently)
• Rate-limiting requests to avoid overwhelming servers
• Tracking and reporting error statistics

Key Concepts Explained

1. Async/Await and Futures

Async/await enables non-blocking I/O without explicit threading or callbacks.

Future: A value that will be available in the future.

#![allow(unused)]
fn main() {
// Synchronous (blocks thread)
fn fetch_sync(url: &str) -> Result<String, Error> {
    // Thread blocks for 2 seconds waiting for response
    std::thread::sleep(Duration::from_secs(2));
    Ok("data".to_string())
}

// Asynchronous (doesn't block)
async fn fetch_async(url: &str) -> Result<String, Error> {
    // Task yields while waiting, thread can do other work
    tokio::time::sleep(Duration::from_secs(2)).await;
    Ok("data".to_string())
}
}

How it works:

• async fn returns a Future (lazy, doesn’t run until .awaited)
• .await yields control to runtime while waiting
• Runtime schedules other tasks on the same thread (cooperative multitasking)

Sync vs Async comparison:

#![allow(unused)]
fn main() {
// Synchronous: 100 requests × 2s = 200 seconds (one thread per request)
for url in urls {
    let data = fetch_sync(url)?;  // Blocks for 2s
}

// Asynchronous: 100 requests = ~2 seconds (all concurrent on one thread)
let futures: Vec<_> = urls.iter().map(|url| fetch_async(url)).collect();
let results = futures::future::join_all(futures).await;  // All run concurrently
}

Why async?

• Memory efficient: 100,000 tasks = ~100MB (vs 100GB for threads)
• Fast: No context switching overhead
• Scalable: Handle millions of concurrent connections

2. Error Handling with thiserror

thiserror simplifies custom error type creation with derive macros.

Without thiserror:

#![allow(unused)]
fn main() {
#[derive(Debug)]
pub enum ScraperError {
    NetworkError(String),
    TimeoutError(u64),
}

// Manual Display implementation (boilerplate!)
impl std::fmt::Display for ScraperError {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        match self {
            ScraperError::NetworkError(msg) => write!(f, "Network error: {}", msg),
            ScraperError::TimeoutError(ms) => write!(f, "Request timed out after {}ms", ms),
        }
    }
}

// Manual Error trait implementation
impl std::error::Error for ScraperError {}
}

With thiserror:

#![allow(unused)]
fn main() {
#[derive(Error, Debug)]
pub enum ScraperError {
    #[error("Network error: {0}")]
    NetworkError(String),

    #[error("Request timed out after {0}ms")]
    TimeoutError(u64),

    #[error("HTTP {status} error for {url}")]
    HttpError { status: u16, url: String },
}
// Done! Display and Error trait auto-implemented
}

Error conversion with From trait:

#![allow(unused)]
fn main() {
impl From<reqwest::Error> for ScraperError {
    fn from(err: reqwest::Error) -> Self {
        if err.is_timeout() {
            ScraperError::TimeoutError(5000)
        } else {
            ScraperError::NetworkError(err.to_string())
        }
    }
}

// Now you can use ? operator
let response = reqwest::get(url).await?;  // Auto-converts reqwest::Error
}

3. Exponential Backoff (Retry Strategy)

Exponential backoff increases delay between retries to avoid overwhelming recovering services.

Linear backoff (BAD):

#![allow(unused)]
fn main() {
// Wait 1s, 1s, 1s, 1s... (doesn't give service time to recover)
for attempt in 0..5 {
    match fetch(url).await {
        Ok(data) => return Ok(data),
        Err(_) => tokio::time::sleep(Duration::from_secs(1)).await,
    }
}
}

Exponential backoff (GOOD):

#![allow(unused)]
fn main() {
// Wait 1s, 2s, 4s, 8s, 16s... (service gets recovery time)
let mut backoff_ms = 1000;
for attempt in 0..5 {
    match fetch(url).await {
        Ok(data) => return Ok(data),
        Err(_) => {
            tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
            backoff_ms *= 2;  // Double each time
        }
    }
}
}

With jitter (prevent thundering herd):

#![allow(unused)]
fn main() {
// Add randomness: 1s±10%, 2s±10%, 4s±10%...
// Prevents all clients retrying at exact same time
let jitter = rand::thread_rng().gen_range(0.9..1.1);
let delay = backoff_ms as f64 * jitter;
tokio::time::sleep(Duration::from_millis(delay as u64)).await;
}

Why exponential backoff?

• Gives services time to recover: Each retry waits longer
• Reduces load: Fewer requests per second over time
• Prevents thundering herd: Jitter spreads retries out

Real-world example: AWS SDK uses exponential backoff with jitter for all API calls.

4. Circuit Breaker Pattern (Fault Tolerance)

Circuit breaker prevents repeated calls to failing services by failing fast.

State machine:

   ┌──────────┐  Too many failures  ┌──────────┐
   │  CLOSED  │────────────────────>│   OPEN   │
   │  (normal)│                     │ (failing)│
   └──────────┘                     └──────────┘
        ^                                  │
        │                                  │ Timeout elapsed
        │  Success                         v
        │                           ┌──────────────┐
        └───────────────────────────│  HALF-OPEN   │
                                    │   (testing)  │
                                    └──────────────┘

Implementation:

#![allow(unused)]
fn main() {
pub enum CircuitState {
    Closed,                      // Normal: allow all requests
    Open { opened_at: Instant }, // Failing: reject immediately
    HalfOpen,                    // Testing: allow one request
}

pub struct CircuitBreaker {
    state: Arc<Mutex<CircuitState>>,
    failure_threshold: usize,    // Failures before opening
    timeout: Duration,           // Time before trying half-open
    consecutive_failures: Arc<Mutex<usize>>,
}
}

How it works:

#![allow(unused)]
fn main() {
async fn call<F>(&self, f: F) -> Result<T, Error> {
    match self.state() {
        CircuitState::Open { opened_at } if opened_at.elapsed() < self.timeout => {
            return Err(Error::CircuitBreakerOpen);  // Fail fast (1ms)
        }
        CircuitState::Open { .. } => {
            self.transition_to(CircuitState::HalfOpen);  // Try again
        }
        _ => {}
    }

    match f.await {
        Ok(result) => {
            self.on_success();  // Reset failure count
            Ok(result)
        }
        Err(e) => {
            self.on_failure();  // Increment, maybe open circuit
            Err(e)
        }
    }
}
}

Impact:

• Without circuit breaker: 1000 requests × 30s timeout = 30,000 seconds wasted
• With circuit breaker: 5 failures → circuit opens → remaining 995 fail in 1ms

5. Concurrency vs Parallelism in Async Rust

Concurrency: Multiple tasks making progress (may run on one thread) Parallelism: Multiple tasks running simultaneously (requires multiple threads)

Tokio async (concurrent, not parallel by default):

#![allow(unused)]
fn main() {
// All run on ONE thread (cooperative multitasking)
tokio::spawn(async { fetch(url1).await });  // Task 1
tokio::spawn(async { fetch(url2).await });  // Task 2
tokio::spawn(async { fetch(url3).await });  // Task 3

// While task 1 waits for network I/O, task 2 runs
// While task 2 waits, task 3 runs, etc.
}

How tokio schedules tasks (simplified):

Thread:  [Task1:fetch] --wait--> [Task2:fetch] --wait--> [Task3:fetch]
                 |                      |                        |
Network:    [I/O pending]         [I/O pending]           [I/O pending]

For CPU-bound work (need parallelism):

#![allow(unused)]
fn main() {
// Spawn on thread pool for CPU-intensive work
tokio::task::spawn_blocking(|| {
    // This runs on separate thread pool
    expensive_computation()
});
}

Concurrent fetching:

#![allow(unused)]
fn main() {
// All 100 requests happen concurrently
let futures: Vec<_> = urls.iter().map(|url| fetch(url)).collect();
let results = join_all(futures).await;  // Wait for all

// Timeline: ~1 second total (limited by slowest request)
// vs sequential: 100 seconds
}

6. Arc and Mutex in Async Context

Arc (Atomic Reference Counting): Share ownership across tasks/threads. Mutex (Mutual Exclusion): Ensure only one task accesses data at a time.

Why Arc?

#![allow(unused)]
fn main() {
let cb = CircuitBreaker::new(3, Duration::from_secs(10));

// ERROR: cb moved into first task, can't use in second
tokio::spawn(async move { cb.call(fetch(url1)).await });
tokio::spawn(async move { cb.call(fetch(url2)).await });  // Error!

// SOLUTION: Arc allows sharing
let cb = Arc::new(CircuitBreaker::new(3, Duration::from_secs(10)));
let cb1 = cb.clone();  // Clone Arc (cheap, just increments counter)
let cb2 = cb.clone();

tokio::spawn(async move { cb1.call(fetch(url1)).await });  // OK
tokio::spawn(async move { cb2.call(fetch(url2)).await });  // OK
}

Why Mutex?

#![allow(unused)]
fn main() {
pub struct CircuitBreaker {
    state: Arc<Mutex<CircuitState>>,  // Multiple tasks need to update state
    consecutive_failures: Arc<Mutex<usize>>,  // Shared counter
}

fn on_failure(&self) {
    let mut failures = self.consecutive_failures.lock().unwrap();
    *failures += 1;  // Safe: only one task can increment at a time
}
}

Async mutex vs std mutex:

#![allow(unused)]
fn main() {
// std::sync::Mutex (use for short critical sections)
let data = self.state.lock().unwrap();  // Blocks thread briefly
*data = new_value;  // Release immediately

// tokio::sync::Mutex (use if holding across .await)
let mut data = self.state.lock().await;  // Yields if contended
some_async_operation().await;  // Can hold lock across await
*data = new_value;
}

When to use each:

• Arc: Share ownership across tasks (always needed for shared state)
• Mutex: Protect mutable shared state
• std::sync::Mutex: Fast, non-async critical sections
• tokio::sync::Mutex: Holding lock across .await points

7. Semaphores (Concurrency Limiting)

Semaphore: Limit number of concurrent operations.

Problem without semaphore:

#![allow(unused)]
fn main() {
// Launch 10,000 requests simultaneously
let futures: Vec<_> = (0..10000).map(|i| fetch(url)).collect();
join_all(futures).await;

// Problems:
// - 10,000 open connections (may exceed OS limits)
// - Huge memory usage (10,000 response buffers)
// - Target server overwhelmed (rate limiting, ban)
}

Solution with semaphore:

#![allow(unused)]
fn main() {
let semaphore = Arc::new(Semaphore::new(100));  // Max 100 concurrent

let futures = urls.iter().map(|url| {
    let semaphore = semaphore.clone();
    async move {
        let _permit = semaphore.acquire().await.unwrap();  // Wait for permit
        fetch(url).await  // Only 100 run at once
        // Permit dropped here, allowing next task to run
    }
});

join_all(futures).await;
}

Visual:

Semaphore with 3 permits:
Task 1 → [Permit 1] → Running
Task 2 → [Permit 2] → Running
Task 3 → [Permit 3] → Running
Task 4 → Waiting...
Task 5 → Waiting...

Task 1 completes → [Permit 1] released
Task 4 → [Permit 1] → Running

Real-world usage:

• Database connection pool: Limit to 10 concurrent queries
• API rate limiting: Max 100 requests/sec
• Resource control: Limit memory-intensive operations

8. Rate Limiting (Token Bucket Algorithm)

Rate limiting controls request rate over time (requests per second).

Token bucket algorithm:

#![allow(unused)]
fn main() {
pub struct RateLimiter {
    tokens: Arc<Mutex<f64>>,         // Available tokens
    rate: f64,                       // Tokens per second
    last_refill: Arc<Mutex<Instant>>, // Last refill time
}

pub async fn acquire(&self) {
    loop {
        let mut tokens = self.tokens.lock().unwrap();
        let mut last = self.last_refill.lock().unwrap();

        // Refill tokens based on elapsed time
        let elapsed = last.elapsed().as_secs_f64();
        *tokens = (*tokens + elapsed * self.rate).min(self.rate);
        *last = Instant::now();

        if *tokens >= 1.0 {
            *tokens -= 1.0;  // Consume token
            break;
        }

        drop(tokens);
        drop(last);

        // Wait before trying again
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
}
}

Visual example (5 requests/second):

Time:     0s    1s    2s    3s
Tokens:   [5] → [4] → [3] → [2] → [1] → [0] → Wait...
Request:   ↓     ↓     ↓     ↓     ↓     ✗
           OK    OK    OK    OK    OK   WAIT

After 1 second: +5 tokens → [5] → Continue

Semaphore vs Rate Limiter:

• Semaphore: Limits concurrent operations (100 at once)
• Rate limiter: Limits rate over time (100 per second)
• Combined: Max 100 concurrent AND max 100/sec

9. Timeout Handling (Bounded Execution Time)

Timeouts ensure operations complete within bounded time.

Without timeout (BAD):

#![allow(unused)]
fn main() {
let response = reqwest::get(url).await?;
// If server never responds, this hangs FOREVER
// Your application is now stuck
}

With timeout (GOOD):

#![allow(unused)]
fn main() {
use tokio::time::{timeout, Duration};

let fetch_future = reqwest::get(url);
match timeout(Duration::from_secs(30), fetch_future).await {
    Ok(Ok(response)) => Ok(response),           // Success
    Ok(Err(e)) => Err(e),                       // Request failed
    Err(_) => Err(Error::Timeout(30000)),       // Timed out
}
}

How tokio::time::timeout works:

#![allow(unused)]
fn main() {
// Simplified implementation
pub async fn timeout<F>(duration: Duration, future: F) -> Result<F::Output, Elapsed> {
    tokio::select! {
        result = future => Ok(result),           // Future completed
        _ = sleep(duration) => Err(Elapsed),     // Timeout fired first
    }
}
}

Timeout at multiple levels:

#![allow(unused)]
fn main() {
// 1. Connection timeout (TCP handshake)
let client = reqwest::Client::builder()
    .connect_timeout(Duration::from_secs(5))
    .build()?;

// 2. Request timeout (entire request/response)
let response = client.get(url)
    .timeout(Duration::from_secs(30))
    .send().await?;

// 3. Application timeout (including retries)
timeout(Duration::from_secs(60), fetch_with_retry(url)).await?;
}

Impact:

• Without timeout: 1 hung request blocks thread forever
• With timeout: Fail after 30s, free resources

10. Partial Results Pattern (Graceful Degradation)

Partial results pattern collects successful results even when some operations fail.

All-or-nothing (BAD for data aggregation):

#![allow(unused)]
fn main() {
// futures::future::try_join_all fails if ANY request fails
let results: Vec<String> = try_join_all(futures).await?;
// If 1 out of 100 fails → you lose ALL 100 results
}

Partial success (GOOD):

#![allow(unused)]
fn main() {
// futures::future::join_all returns all results (Ok or Err)
let results: Vec<Result<String, Error>> = join_all(futures).await;

// Extract successes and failures
let successes: Vec<String> = results.iter()
    .filter_map(|r| r.as_ref().ok())
    .collect();

let failures: Vec<&Error> = results.iter()
    .filter_map(|r| r.as_ref().err())
    .collect();

println!("Success: {}/100, Failed: {}/100", successes.len(), failures.len());
// Output: "Success: 95/100, Failed: 5/100" → 95% data recovered!
}

Structured partial results:

#![allow(unused)]
fn main() {
pub struct FetchResult {
    pub url: String,
    pub result: Result<String, ScraperError>,
    pub duration_ms: u64,
    pub attempt_count: usize,
}

pub struct FetchSummary {
    pub total: usize,
    pub success: usize,
    pub failed: usize,
    pub avg_duration_ms: u64,
}
}

When to use:

• ✅ Data aggregation: Scraping 1000 websites (some will fail)
• ✅ Health checks: Ping 100 servers (want status of all)
• ✅ Batch processing: Process 10,000 images (continue on errors)
• ❌ Transactions: Bank transfer (must succeed atomically)
• ❌ Critical operations: Deploy code (all or nothing)

Connection to This Project

This project builds production-ready async web scraper with sophisticated error handling and resilience patterns.

Milestone 1: Basic Async HTTP Client with Error Types

Concepts applied:

• Async/await with tokio runtime
• Custom error types with thiserror
• Error conversion with From trait
• Result type for error propagation

Why it matters: Async I/O is essential for scalable network applications:

• 1 thread can handle 10,000+ concurrent connections
• No blocking on network I/O (cooperative multitasking)
• Memory efficient (100KB per task vs 2MB per thread)

Real-world impact:

#![allow(unused)]
fn main() {
// Synchronous (blocking): 100 URLs = 100 threads = 200MB memory
for url in urls {
    let response = blocking_fetch(url)?;  // Blocks thread for ~2s
}
// Total time: 100 × 2s = 200 seconds

// Asynchronous (non-blocking): 100 URLs = 1 thread = 2MB memory
let futures = urls.iter().map(|url| fetch_url(url));
let results = join_all(futures).await;
// Total time: ~2 seconds (all concurrent)
}

Performance comparison:

Milestone 2: Add Timeout to Prevent Hanging

Concepts applied:

• Timeout with tokio::time::timeout
• Bounded execution time for all operations
• Timeout error variant

Why it matters: Without timeouts, one slow server hangs your application indefinitely:

• Blocks thread/task forever
• Resource exhaustion (100 hung requests = 100 blocked threads)
• Cascade failures (dependent services time out waiting for you)

Real-world impact:

#![allow(unused)]
fn main() {
// Without timeout: Server never responds → hang forever
let response = fetch_url(url).await?;  // Stuck here forever

// With timeout: Fail after 30s, continue with other work
match timeout(Duration::from_secs(30), fetch_url(url)).await {
    Ok(Ok(resp)) => Ok(resp),
    Ok(Err(e)) => Err(e),
    Err(_) => Err(TimeoutError(30000)),  // Fail fast after 30s
}
}

Impact comparison:

Real-world validation: AWS SDK sets 30-60s timeouts on all operations, preventing hung connections.

Milestone 3: Retry with Exponential Backoff

Concepts applied:

• Exponential backoff algorithm
• Retry logic with configurable attempts
• Jitter to prevent thundering herd
• Error classification (retryable vs permanent)

Why it matters: Network failures are often transient (temporary):

• DNS timeout (resolves in 1s)
• Connection refused (server restarting, ready in 5s)
• 503 Service Unavailable (server overloaded, recovers in 10s)

Exponential backoff transforms “50% failure rate” into “99% success rate”.

Real-world impact:

#![allow(unused)]
fn main() {
// No retry: 50% transient failure rate → 50% success
let result = fetch_url(url).await?;

// With retry (3 attempts): → 99% success rate
// Attempt 1: 50% fail → Retry after 1s
// Attempt 2: 50% of 50% = 25% fail → Retry after 2s
// Attempt 3: 50% of 25% = 12.5% fail → Retry after 4s
// Final success rate: 1 - (0.5^3) = 87.5% → with jitter ~99%
}

Success rate by attempts:

Jitter prevents thundering herd:

• Without jitter: 1000 clients retry at exact same time → overwhelm server
• With jitter: Retries spread over 900ms-1100ms → smooth load

Milestone 4: Circuit Breaker Pattern

Concepts applied:

• State machine (Closed/Open/HalfOpen)
• Arc and Mutex for shared state across tasks
• Fail-fast when service is known to be down
• Automatic recovery testing (half-open state)

Why it matters: Retries help with transient failures, but if a service is completely down:

• Every retry waits for full timeout (30s per request)
• 1000 retries × 30s = 30,000 seconds of wasted time
• Resources exhausted (threads, memory, connections)
• Cascading failures (your service becomes slow, downstream services timeout)

Circuit breaker fails fast (1ms) instead of waiting (30s).

Real-world impact:

#![allow(unused)]
fn main() {
// Without circuit breaker: Service down, 1000 requests
// 1000 requests × 30s timeout = 30,000 seconds wasted
// 1000 threads blocked waiting

// With circuit breaker:
// Request 1-5: Fail after 30s each (threshold = 5) → 150s
// Circuit opens
// Request 6-1000: Fail immediately in 1ms → 1 second
// Total: 150s + 1s = 151s (vs 30,000s)
}

Performance comparison (service completely down, 1000 requests):

Real-world validation:

• Netflix Hystrix: Circuit breaker for microservices (saved millions in downtime)
• AWS SDK: Built-in circuit breaker for API calls
• Kubernetes: Circuit breaking in service mesh (Istio)

Milestone 5: Concurrent Fetching with Partial Results

Concepts applied:

• Concurrency with join_all
• Partial results pattern (graceful degradation)
• Result aggregation and statistics
• Structured error reporting

Why it matters: Sequential fetching is slow, “all-or-nothing” is fragile:

• Sequential: 100 URLs × 1s = 100 seconds
• All-or-nothing: 1 failure → lose all 100 results

Concurrent + partial results: 100 URLs in ~1s, get 95 results if 5 fail.

Real-world impact:

#![allow(unused)]
fn main() {
// Sequential: 100 seconds for 100 URLs
for url in urls {
    let result = fetch(url).await?;  // 1s each, sequential
}

// Concurrent: ~1 second for 100 URLs
let futures = urls.iter().map(|url| fetch(url));
let results = join_all(futures).await;  // All run concurrently
}

Performance comparison (100 URLs, 1s each):

Real-world scenarios:

• Web scraping: Scrape 10,000 websites (5% will fail) → get 9,500 results
• Health monitoring: Check 1,000 servers → status report even if some timeout
• Data aggregation: Fetch from 50 APIs → combine all available data

Milestone 6: Rate Limiting and Resource Management

Concepts applied:

• Semaphores for concurrency limiting
• Rate limiting with token bucket
• Resource tracking (memory, connections)
• Production-ready error handling

Why it matters: Unlimited concurrency causes:

• Client-side: OOM (out of memory), file descriptor exhaustion
• Server-side: Rate limiting (429 errors), IP bans, degraded performance

Rate limiting is respectful (doesn’t DDoS targets) and prevents resource exhaustion.

Real-world impact:

#![allow(unused)]
fn main() {
// Without limits: Launch 10,000 requests immediately
let futures: Vec<_> = urls.iter().map(|url| fetch(url)).collect();
join_all(futures).await;  // 10,000 concurrent connections
// Result: OOM crash, IP banned, server overwhelmed

// With limits: Max 100 concurrent, max 100/sec
let semaphore = Arc::new(Semaphore::new(100));
let rate_limiter = RateLimiter::new(100, 100.0);

let futures = urls.iter().map(|url| {
    let sem = semaphore.clone();
    let limiter = rate_limiter.clone();
    async move {
        let _permit = sem.acquire().await;     // Wait for concurrency slot
        let _token = limiter.acquire().await;  // Wait for rate limit token
        fetch(url).await
    }
});
// Result: Stable, respectful, no resource exhaustion
}

Resource usage comparison (10,000 requests):

Real-world validation:

• GitHub API: 5,000 requests/hour limit
• Twitter API: 300 requests/15min window
• Production scrapers: Always use rate limiting (respectful, prevents bans)

Milestone 1: Basic Async HTTP Client with Error Types

Goal: Create async HTTP client with typed errors using reqwest and tokio.

What to implement:

• Define ScraperError enum with variants for different failure modes
• Implement async fetch_url() function using reqwest
• Convert reqwest errors to your error type
• Basic error handling with Result types

Architecture:

• Enums: ScraperError
• Functions:
  • fetch_url(url: &str) -> Result<String, ScraperError> - Async HTTP GET
  • Error conversion from reqwest errors

Starter Code:

#![allow(unused)]
fn main() {
use reqwest;
use thiserror::Error;
use tokio;

/// Comprehensive error type for web scraping
/// Role: Typed error handling for network operations
#[derive(Error, Debug, Clone)]
pub enum ScraperError {
    #[error("Network error: {0}")]
    NetworkError(String),                                   // Connection failures, DNS errors   
                                                                 
    #[error("Request timed out after {0}ms")]                     
    TimeoutError(u64),                                      // Request exceeded time limit        
                                                            
    #[error("HTTP {status} error for {url}")]               
    HttpError { status: u16, url: String },                 // Non-success HTTP status codes  
                                                              
    #[error("Failed to parse response: {0}")]                
    ParseError(String),                                     // Failed to parse response body 
}

/// Fetch URL content
/// Role: Basic async HTTP GET request
pub async fn fetch_url(url: &str) -> Result<String, ScraperError> {
    todo!("Implement async HTTP GET with error conversion")
}

/// Convert reqwest errors to ScraperError
impl From<reqwest::Error> for ScraperError {
    /// Role: Map reqwest errors to our error type
    fn from(err: reqwest::Error) -> Self {
        todo!("Convert based on error type - timeout, connection, etc.")
    }
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;
    use wiremock::{Mock, MockServer, ResponseTemplate};
    use wiremock::matchers::{method, path};

    #[tokio::test]
    async fn test_fetch_url_success() {
        let mock_server = MockServer::start().await;

        Mock::given(method("GET"))
            .and(path("/test"))
            .respond_with(ResponseTemplate::new(200).set_body_string("Hello, World!"))
            .mount(&mock_server)
            .await;

        let url = format!("{}/test", &mock_server.uri());
        let result = fetch_url(&url).await;

        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "Hello, World!");
    }

    #[tokio::test]
    async fn test_fetch_url_404() {
        let mock_server = MockServer::start().await;

        Mock::given(method("GET"))
            .and(path("/notfound"))
            .respond_with(ResponseTemplate::new(404))
            .mount(&mock_server)
            .await;

        let url = format!("{}/notfound", &mock_server.uri());
        let result = fetch_url(&url).await;

        assert!(result.is_err());
        match result.unwrap_err() {
            ScraperError::HttpError { status, .. } => assert_eq!(status, 404),
            _ => panic!("Expected HttpError"),
        }
    }

    #[tokio::test]
    async fn test_fetch_url_network_error() {
        // Invalid URL
        let result = fetch_url("http://invalid.local.test").await;
        assert!(result.is_err());
        assert!(matches!(result.unwrap_err(), ScraperError::NetworkError(_)));
    }

    #[tokio::test]
    async fn test_error_display() {
        let error = ScraperError::HttpError {
            status: 500,
            url: "http://example.com".to_string(),
        };

        let display = format!("{}", error);
        assert!(display.contains("500"));
        assert!(display.contains("example.com"));
    }

    #[tokio::test]
    async fn test_error_is_send_and_sync() {
        fn assert_send<T: Send>() {}
        fn assert_sync<T: Sync>() {}

        assert_send::<ScraperError>();
        assert_sync::<ScraperError>();
    }
}
}

Milestone 2: Add Timeout to Prevent Hanging

Goal: Ensure all network operations have bounded execution time using tokio::time::timeout.

Why the previous milestone is not enough: Without timeouts, slow or hanging servers can block your application indefinitely, causing resource exhaustion. One stuck request can hold threads/tasks forever, preventing progress on other work.

What’s the improvement: Timeouts guarantee bounded waiting time. If a server takes >30s, you fail fast and move on. This prevents one slow endpoint from blocking 100 other requests. Essential for responsive systems. The difference is between “hangs forever” and “fails after 30s and continues”.

Architecture:

• Functions:
  • fetch_url_with_timeout(url: &str, timeout_ms: u64) -> Result<String, ScraperError> - Fetch with timeout
  • Use tokio::time::timeout() wrapper

Starter Code:

#![allow(unused)]
fn main() {
use tokio::time::{timeout, Duration};

/// Fetch URL with timeout
/// Role: Bounded execution time for network operations
pub async fn fetch_url_with_timeout(
    url: &str,
    timeout_ms: u64,
) -> Result<String, ScraperError> {
    todo!("Wrap fetch_url in tokio::time::timeout")
}

/// Create HTTP client with timeout
/// Role: Configure reqwest client with timeout
pub fn create_client(timeout_ms: u64) -> reqwest::Client {
    todo!("Build reqwest::Client with timeout configuration")
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;
    use wiremock::{Mock, MockServer, ResponseTemplate};
    use tokio::time::Duration;

    #[tokio::test]
    async fn test_timeout_success_within_limit() {
        let mock_server = MockServer::start().await;

        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200).set_body_string("Quick response"))
            .mount(&mock_server)
            .await;

        let result = fetch_url_with_timeout(&mock_server.uri(), 5000).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_timeout_triggers() {
        let mock_server = MockServer::start().await;

        // Delay response by 2 seconds
        Mock::given(method("GET"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_delay(Duration::from_secs(2))
            )
            .mount(&mock_server)
            .await;

        // Timeout after 100ms
        let result = fetch_url_with_timeout(&mock_server.uri(), 100).await;

        assert!(result.is_err());
        match result.unwrap_err() {
            ScraperError::TimeoutError(ms) => assert_eq!(ms, 100),
            e => panic!("Expected TimeoutError, got {:?}", e),
        }
    }

    #[tokio::test]
    async fn test_client_timeout_configuration() {
        let client = create_client(1000);

        // Client should have timeout configured
        // This is integration-tested through fetch_url_with_timeout
    }

    #[tokio::test]
    async fn test_timeout_error_includes_duration() {
        let error = ScraperError::TimeoutError(5000);
        let display = format!("{}", error);
        assert!(display.contains("5000"));
        assert!(display.contains("timed out"));
    }
}
}

Milestone 3: Retry with Exponential Backoff

Goal: Automatically retry failed requests with increasing delays and jitter.

Why the previous milestone is not enough: Network errors are often transient (temporary DNS issues, brief connection drops). A single retry could succeed, but immediate retry might fail again if the issue needs time to resolve. We need smart retry strategy.

What’s the improvement: Exponential backoff gives services time to recover (1s, 2s, 4s, 8s…). Jitter prevents thundering herd (synchronized retries from many clients overwhelming the recovering server). This transforms “50% of requests fail due to transient errors” into “99% success rate with retries”. The 1-2% permanent failures are caught appropriately.

Optimization focus: Reliability through intelligent retry strategy - maximize success rate while respecting server recovery time.

Architecture:

• Structs: RetryConfig
• Functions:
  • fetch_with_retry(url, timeout, config) -> Result<String, ScraperError> - Retry logic
  • ScraperError::is_retryable() -> bool - Check if error warrants retry
  • calculate_backoff(attempt: usize) -> Duration - Exponential backoff with jitter

Starter Code:

#![allow(unused)]
fn main() {
use rand::Rng;

///  Configure retry behavior
#[derive(Debug, Clone)]
pub struct RetryConfig {
    pub max_attempts: usize,                // Maximum retry attempts              
    pub initial_backoff_ms: u64,            // Starting backoff delay          
    pub max_backoff_ms: u64,                // Maximum backoff delay               
    pub jitter: bool,                       // Randomness to prevent thundering herd  
}

impl RetryConfig {
    /// Create default retry config
    /// Role: Sensible defaults for most cases
    pub fn default() -> Self {
        todo!("Return config with 3 attempts, 1s initial, 30s max")
    }

    /// Calculate backoff for attempt
    /// Role: Exponential backoff with optional jitter
    pub fn backoff_duration(&self, attempt: usize) -> Duration {
        todo!("Calculate 2^attempt * initial, apply jitter if enabled")
    }
}

impl ScraperError {
    /// Check if error is retryable
    /// Role: Classify errors as transient or permanent
    pub fn is_retryable(&self) -> bool {
        todo!("Return true for network errors, timeouts, 5xx; false for 4xx")
    }
}

/// Fetch URL with retry logic
/// Role: Resilient HTTP requests
pub async fn fetch_with_retry(
    url: &str,
    timeout_ms: u64,
    retry_config: &RetryConfig,
) -> Result<String, ScraperError> {
    todo!("Loop up to max_attempts, apply backoff between retries")
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::{Arc, Mutex};
    use std::time::Instant;

    #[tokio::test]
    async fn test_retry_succeeds_after_failures() {
        let mock_server = MockServer::start().await;
        let attempt_count = Arc::new(Mutex::new(0));
        let attempt_count_clone = attempt_count.clone();

        // Fail first 2 times, succeed on 3rd
        Mock::given(method("GET"))
            .respond_with(move |_req: &wiremock::Request| {
                let mut count = attempt_count_clone.lock().unwrap();
                *count += 1;

                if *count < 3 {
                    ResponseTemplate::new(500)
                } else {
                    ResponseTemplate::new(200).set_body_string("Success!")
                }
            })
            .mount(&mock_server)
            .await;

        let config = RetryConfig {
            max_attempts: 3,
            initial_backoff_ms: 10,
            max_backoff_ms: 1000,
            jitter: false,
        };

        let result = fetch_with_retry(&mock_server.uri(), 1000, &config).await;

        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "Success!");
        assert_eq!(*attempt_count.lock().unwrap(), 3);
    }

    #[tokio::test]
    async fn test_retry_gives_up_after_max_attempts() {
        let mock_server = MockServer::start().await;

        // Always fail
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(500))
            .mount(&mock_server)
            .await;

        let config = RetryConfig {
            max_attempts: 3,
            initial_backoff_ms: 10,
            max_backoff_ms: 1000,
            jitter: false,
        };

        let result = fetch_with_retry(&mock_server.uri(), 1000, &config).await;

        assert!(result.is_err());
        match result.unwrap_err() {
            ScraperError::HttpError { status, .. } => assert_eq!(status, 500),
            e => panic!("Expected HttpError, got {:?}", e),
        }
    }

    #[tokio::test]
    async fn test_no_retry_on_404() {
        let mock_server = MockServer::start().await;
        let attempt_count = Arc::new(Mutex::new(0));
        let attempt_count_clone = attempt_count.clone();

        Mock::given(method("GET"))
            .respond_with(move |_req: &wiremock::Request| {
                let mut count = attempt_count_clone.lock().unwrap();
                *count += 1;
                ResponseTemplate::new(404)
            })
            .mount(&mock_server)
            .await;

        let config = RetryConfig {
            max_attempts: 3,
            initial_backoff_ms: 10,
            max_backoff_ms: 1000,
            jitter: false,
        };

        let result = fetch_with_retry(&mock_server.uri(), 1000, &config).await;

        assert!(result.is_err());
        // Should only try once (404 is not retryable)
        assert_eq!(*attempt_count.lock().unwrap(), 1);
    }

    #[test]
    fn test_exponential_backoff() {
        let config = RetryConfig {
            max_attempts: 5,
            initial_backoff_ms: 100,
            max_backoff_ms: 10000,
            jitter: false,
        };

        assert_eq!(config.backoff_duration(0).as_millis(), 100);
        assert_eq!(config.backoff_duration(1).as_millis(), 200);
        assert_eq!(config.backoff_duration(2).as_millis(), 400);
        assert_eq!(config.backoff_duration(3).as_millis(), 800);
    }

    #[test]
    fn test_backoff_respects_max() {
        let config = RetryConfig {
            max_attempts: 10,
            initial_backoff_ms: 100,
            max_backoff_ms: 1000,
            jitter: false,
        };

        // Should cap at max_backoff_ms
        assert!(config.backoff_duration(10).as_millis() <= 1000);
    }

    #[test]
    fn test_jitter_adds_randomness() {
        let config = RetryConfig {
            max_attempts: 3,
            initial_backoff_ms: 100,
            max_backoff_ms: 10000,
            jitter: true,
        };

        let d1 = config.backoff_duration(1);
        let d2 = config.backoff_duration(1);

        // With jitter, same attempt should give different durations
        // This is probabilistic but very likely
        // Note: May rarely fail due to randomness
    }

    #[tokio::test]
    async fn test_retry_timing() {
        let mock_server = MockServer::start().await;

        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(500))
            .mount(&mock_server)
            .await;

        let config = RetryConfig {
            max_attempts: 3,
            initial_backoff_ms: 100,
            max_backoff_ms: 10000,
            jitter: false,
        };

        let start = Instant::now();
        let _ = fetch_with_retry(&mock_server.uri(), 1000, &config).await;
        let elapsed = start.elapsed();

        // Should take at least 100ms + 200ms = 300ms for backoffs
        assert!(elapsed.as_millis() >= 300);
    }

    #[test]
    fn test_error_retryability() {
        assert!(ScraperError::NetworkError("test".to_string()).is_retryable());
        assert!(ScraperError::TimeoutError(1000).is_retryable());
        assert!(ScraperError::HttpError { status: 500, url: "".to_string() }.is_retryable());
        assert!(ScraperError::HttpError { status: 503, url: "".to_string() }.is_retryable());

        assert!(!ScraperError::HttpError { status: 404, url: "".to_string() }.is_retryable());
        assert!(!ScraperError::HttpError { status: 403, url: "".to_string() }.is_retryable());
        assert!(!ScraperError::ParseError("test".to_string()).is_retryable());
    }
}
}

Milestone 4: Circuit Breaker Pattern

Goal: Prevent repeated calls to failing services using circuit breaker state machine.

Why the previous milestone is not enough: Retries help with transient failures, but if a service is completely down, retrying wastes time and resources. Every retry waits for timeout, consuming threads and memory. A failing service receiving constant retries can’t recover.

What’s the improvement: Circuit breakers fail fast when a service is known to be down. Instead of waiting 30s for timeout on every request, fail in 1ms. This prevents cascading failures: if service A is down, service B (which depends on A) stops hammering it with requests, allowing A to recover. Resource exhaustion is prevented, latency drops dramatically (1ms vs 30s), and failing services get breathing room to recover.

Optimization focus: Speed (fail fast) and reliability (allow service recovery).

Architecture:

• Enums: CircuitState (Closed, Open, HalfOpen)
• Structs: CircuitBreaker
• Fields: state: Arc<Mutex<CircuitState>>, failure_count: Arc<Mutex<usize>>, threshold: usize
• Functions:
  • CircuitBreaker::new(threshold, timeout) - Create circuit breaker
  • call<F>(&self, f: F) -> Result<T, ScraperError> - Execute with circuit breaker
  • State transition logic (closed → open → half-open → closed)

Starter Code:

#![allow(unused)]
fn main() {
use std::sync::{Arc, Mutex};
use std::time::{Instant, Duration};
use std::future::Future;

/// Circuit breaker states
/// Role: Track service health
#[derive(Debug, Clone, PartialEq)]
pub enum CircuitState {
    Closed,                                   // Normal operation, requests pass through       
    Open { opened_at: Instant },              // Service failing, reject requests immediately    
    HalfOpen,                                 // Testing if service recovered                
}

/// Circuit breaker for preventing cascading failures
/// Fault tolerance component
pub struct CircuitBreaker {
    state: Arc<Mutex<CircuitState>>,                      // Current state              
    failure_threshold: usize,                             // Failures before opening           
    success_threshold: usize,                             // Successes to close from half-open 
    timeout: Duration,                                    // Time before trying half-open             
    consecutive_failures: Arc<Mutex<usize>>,              // Failure counter    
    consecutive_successes: Arc<Mutex<usize>>,             // Success counter   
}

impl CircuitBreaker {
    /// Create new circuit breaker
    /// Role: Initialize with thresholds and timeout
    pub fn new(failure_threshold: usize, timeout: Duration) -> Self {
        todo!("Initialize circuit breaker in Closed state")
    }

    /// Execute function with circuit breaker protection
    /// Role: Wrap async call with circuit breaker logic
    pub async fn call<F, T>(&self, f: F) -> Result<T, ScraperError>
    where
        F: Future<Output = Result<T, ScraperError>>,
    {
        todo!("Check state, execute if closed/half-open, update state based on result")
    }

    /// Handle successful request
    /// Role: Update state after success
    fn on_success(&self) {
        todo!("Reset failure count, increment success count, transition to Closed if needed")
    }

    /// Handle failed request
    /// Role: Update state after failure
    fn on_failure(&self) {
        todo!("Increment failure count, transition to Open if threshold exceeded")
    }

    /// Get current state
    /// Role: Query circuit breaker state
    pub fn state(&self) -> CircuitState {
        todo!("Return current state")
    }

    /// Check if should attempt request
    /// Role: Determine if circuit allows request
    fn should_attempt(&self) -> bool {
        todo!("Return false if Open and timeout not elapsed")
    }
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_circuit_breaker_opens_after_failures() {
        let cb = CircuitBreaker::new(3, Duration::from_secs(1));

        assert_eq!(cb.state(), CircuitState::Closed);

        // Cause 3 failures
        for _ in 0..3 {
            let result = cb.call(async {
                Err::<(), _>(ScraperError::NetworkError("fail".to_string()))
            }).await;
            assert!(result.is_err());
        }

        // Circuit should now be open
        match cb.state() {
            CircuitState::Open { .. } => {},
            state => panic!("Expected Open, got {:?}", state),
        }
    }

    #[tokio::test]
    async fn test_circuit_breaker_fails_fast_when_open() {
        let cb = CircuitBreaker::new(2, Duration::from_secs(1));

        // Open the circuit
        for _ in 0..2 {
            let _ = cb.call(async {
                Err::<(), _>(ScraperError::NetworkError("fail".to_string()))
            }).await;
        }

        // Next call should fail immediately without executing
        let start = Instant::now();
        let result = cb.call(async {
            tokio::time::sleep(Duration::from_secs(1)).await;
            Ok::<_, ScraperError>(())
        }).await;
        let elapsed = start.elapsed();

        assert!(result.is_err());
        // Should fail instantly (< 100ms), not wait for sleep
        assert!(elapsed.as_millis() < 100);
    }

    #[tokio::test]
    async fn test_circuit_breaker_transitions_to_half_open() {
        let cb = CircuitBreaker::new(2, Duration::from_millis(100));

        // Open the circuit
        for _ in 0..2 {
            let _ = cb.call(async {
                Err::<(), _>(ScraperError::NetworkError("fail".to_string()))
            }).await;
        }

        // Wait for timeout
        tokio::time::sleep(Duration::from_millis(150)).await;

        // Next call should transition to HalfOpen
        let _ = cb.call(async { Ok::<_, ScraperError>(()) }).await;

        assert_eq!(cb.state(), CircuitState::HalfOpen);
    }

    #[tokio::test]
    async fn test_circuit_breaker_closes_after_successes() {
        let cb = CircuitBreaker::new(2, Duration::from_millis(100));

        // Open the circuit
        for _ in 0..2 {
            let _ = cb.call(async {
                Err::<(), _>(ScraperError::NetworkError("fail".to_string()))
            }).await;
        }

        // Wait and succeed twice (assuming success_threshold = 2)
        tokio::time::sleep(Duration::from_millis(150)).await;

        for _ in 0..2 {
            let result = cb.call(async { Ok::<_, ScraperError>(()) }).await;
            assert!(result.is_ok());
        }

        // Circuit should be closed
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[tokio::test]
    async fn test_circuit_breaker_reopens_on_half_open_failure() {
        let cb = CircuitBreaker::new(2, Duration::from_millis(100));

        // Open the circuit
        for _ in 0..2 {
            let _ = cb.call(async {
                Err::<(), _>(ScraperError::NetworkError("fail".to_string()))
            }).await;
        }

        // Wait for timeout
        tokio::time::sleep(Duration::from_millis(150)).await;

        // Try in half-open and fail
        let _ = cb.call(async {
            Err::<(), _>(ScraperError::NetworkError("still failing".to_string()))
        }).await;

        // Should reopen
        match cb.state() {
            CircuitState::Open { .. } => {},
            state => panic!("Expected Open, got {:?}", state),
        }
    }

    #[tokio::test]
    async fn test_circuit_breaker_success_resets_failure_count() {
        let cb = CircuitBreaker::new(3, Duration::from_secs(1));

        // Two failures
        for _ in 0..2 {
            let _ = cb.call(async {
                Err::<(), _>(ScraperError::NetworkError("fail".to_string()))
            }).await;
        }

        // One success
        let _ = cb.call(async { Ok::<_, ScraperError>(()) }).await;

        // Another failure shouldn't open (count was reset)
        let _ = cb.call(async {
            Err::<(), _>(ScraperError::NetworkError("fail".to_string()))
        }).await;

        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[tokio::test]
    async fn test_circuit_breaker_with_real_fetch() {
        let mock_server = MockServer::start().await;
        let cb = Arc::new(CircuitBreaker::new(2, Duration::from_millis(100)));

        // Setup failing endpoint
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(500))
            .mount(&mock_server)
            .await;

        let url = mock_server.uri();

        // Open circuit with failures
        for _ in 0..2 {
            let cb_clone = cb.clone();
            let url_clone = url.clone();
            let _ = cb_clone.call(fetch_url(&url_clone)).await;
        }

        // Verify circuit is open
        match cb.state() {
            CircuitState::Open { .. } => {},
            state => panic!("Expected Open, got {:?}", state),
        }
    }
}
}

Milestone 5: Concurrent Fetching with Partial Results

Goal: Fetch multiple URLs concurrently and collect partial results even if some fail.

Why the previous milestone is not enough: Sequential fetching is slow. Fetching 100 URLs at 1 second each takes 100 seconds. We have retry logic and circuit breakers, but they don’t help with the sequential bottleneck. Also, naive try_join_all fails completely if ANY request fails, losing all successful data.

What’s the improvement: Parallelism provides massive speedup for I/O-bound operations - 100 concurrent requests take ~1 second (limited by slowest). Collecting partial results means “get as much data as possible” rather than “all or nothing”. For data aggregation where 95/100 URLs succeed, you get 95% of the data instead of 0%. This is essential for resilient data pipelines.

Optimization focus: Speed through parallelism and resilience through partial success collection.

Architecture:

• Structs: FetchResult, FetchSummary
• Functions:
  • fetch_all(urls, config) -> Vec<FetchResult> - Concurrent fetching
  • FetchSummary::from_results() - Aggregate statistics

Starter Code:

#![allow(unused)]
fn main() {
use futures::future::join_all;
use std::time::Instant;

/// Result of fetching a single URL
/// Role: Record individual fetch outcomes
#[derive(Debug)]
pub struct FetchResult {
    pub url: String,                                      // The URL that was fetched                       
    pub result: Result<String, ScraperError>,             // Success or failure    
    pub duration_ms: u64,                                 // Time taken for this fetch                 
    pub attempt_count: usize,                             // Number of retries needed              
}

impl FetchResult {
    /// Check if fetch succeeded
    pub fn is_success(&self) -> bool {
        todo!("Query success status")
    }

    /// Get content if successful
    pub fn content(&self) -> Option<&str> {
        todo!("Extract content from result")
    }
}

/// FetchSummary: Provide overview of batch operation
#[derive(Debug)]
pub struct FetchSummary {
    pub total: usize,                        // Total URLs attempted              
    pub success: usize,                      // Successful fetches              
    pub failed: usize,                       // Failed fetches                   
    pub total_duration_ms: u64,              // Total time for batch    
    pub avg_duration_ms: u64,                // Average time per fetch    
}

impl FetchSummary {
    /// Create summary from results
    /// Role: Aggregate statistics
    pub fn from_results(results: &[FetchResult]) -> Self {
        todo!("Calculate success/failure counts, timing statistics")
    }

    /// Calculate success rate as percentage
    pub fn success_rate(&self) -> f64 {
      todo!("Compute reliability metric")
    }
}

/// Fetch multiple URLs concurrently
/// Role: Maximize throughput with parallel I/O
pub async fn fetch_all(
    urls: Vec<String>,
    timeout_ms: u64,
    retry_config: &RetryConfig,
    circuit_breaker: &Arc<CircuitBreaker>,
) -> Vec<FetchResult> {
    todo!("Create futures for all URLs, use join_all to execute concurrently")
}

/// Fetch URLs with rate limiting
/// Role: Control resource usage
pub async fn fetch_all_with_limit(
    urls: Vec<String>,
    timeout_ms: u64,
    retry_config: &RetryConfig,
    circuit_breaker: &Arc<CircuitBreaker>,
    max_concurrent: usize,
) -> Vec<FetchResult> {
    todo!("Use semaphore to limit concurrent requests")
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_fetch_all_concurrent() {
        let mock_server = MockServer::start().await;

        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200).set_body_string("data"))
            .mount(&mock_server)
            .await;

        let urls = vec![
            format!("{}/1", mock_server.uri()),
            format!("{}/2", mock_server.uri()),
            format!("{}/3", mock_server.uri()),
        ];

        let config = RetryConfig::default();
        let cb = Arc::new(CircuitBreaker::new(5, Duration::from_secs(10)));

        let start = Instant::now();
        let results = fetch_all(urls, 5000, &config, &cb).await;
        let elapsed = start.elapsed();

        // All should succeed
        assert_eq!(results.len(), 3);
        assert!(results.iter().all(|r| r.is_success()));

        // Should be concurrent (not 3x sequential time)
        // Hard to assert precisely, but should be reasonably fast
        assert!(elapsed.as_secs() < 2);
    }

    #[tokio::test]
    async fn test_fetch_all_partial_success() {
        let mock_server = MockServer::start().await;

        // First endpoint succeeds
        Mock::given(method("GET"))
            .and(path("/good"))
            .respond_with(ResponseTemplate::new(200).set_body_string("success"))
            .mount(&mock_server)
            .await;

        // Second endpoint fails
        Mock::given(method("GET"))
            .and(path("/bad"))
            .respond_with(ResponseTemplate::new(500))
            .mount(&mock_server)
            .await;

        let urls = vec![
            format!("{}/good", mock_server.uri()),
            format!("{}/bad", mock_server.uri()),
        ];

        let config = RetryConfig { max_attempts: 1, ..RetryConfig::default() };
        let cb = Arc::new(CircuitBreaker::new(5, Duration::from_secs(10)));

        let results = fetch_all(urls, 5000, &config, &cb).await;

        assert_eq!(results.len(), 2);
        assert!(results[0].is_success());
        assert!(!results[1].is_success());
    }

    #[test]
    fn test_fetch_summary_calculations() {
        let results = vec![
            FetchResult {
                url: "url1".to_string(),
                result: Ok("data".to_string()),
                duration_ms: 100,
                attempt_count: 1,
            },
            FetchResult {
                url: "url2".to_string(),
                result: Err(ScraperError::NetworkError("fail".to_string())),
                duration_ms: 200,
                attempt_count: 3,
            },
            FetchResult {
                url: "url3".to_string(),
                result: Ok("data".to_string()),
                duration_ms: 150,
                attempt_count: 1,
            },
        ];

        let summary = FetchSummary::from_results(&results);

        assert_eq!(summary.total, 3);
        assert_eq!(summary.success, 2);
        assert_eq!(summary.failed, 1);
        assert_eq!(summary.success_rate(), 66.66666666666666);
    }

    #[tokio::test]
    async fn test_fetch_all_with_concurrency_limit() {
        let mock_server = MockServer::start().await;

        Mock::given(method("GET"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_delay(Duration::from_millis(100))
            )
            .mount(&mock_server)
            .await;

        let urls: Vec<String> = (0..10)
            .map(|i| format!("{}/{}", mock_server.uri(), i))
            .collect();

        let config = RetryConfig::default();
        let cb = Arc::new(CircuitBreaker::new(10, Duration::from_secs(10)));

        let start = Instant::now();
        let results = fetch_all_with_limit(urls, 5000, &config, &cb, 3).await;
        let elapsed = start.elapsed();

        assert_eq!(results.len(), 10);

        // With limit of 3 and 100ms delay, should take roughly:
        // 10 requests / 3 concurrent = ~4 batches * 100ms = ~400ms
        assert!(elapsed.as_millis() >= 300);
        assert!(elapsed.as_millis() < 600);
    }

    #[tokio::test]
    async fn test_fetch_result_methods() {
        let success = FetchResult {
            url: "http://example.com".to_string(),
            result: Ok("content".to_string()),
            duration_ms: 100,
            attempt_count: 1,
        };

        assert!(success.is_success());
        assert_eq!(success.content(), Some("content"));

        let failure = FetchResult {
            url: "http://example.com".to_string(),
            result: Err(ScraperError::NetworkError("fail".to_string())),
            duration_ms: 100,
            attempt_count: 3,
        };

        assert!(!failure.is_success());
        assert_eq!(failure.content(), None);
    }
}
}

Milestone 6: Rate Limiting and Resource Management

Goal: Add rate limiting to avoid overwhelming servers and manage client resources.

Why the previous milestone is not enough: Unlimited concurrent requests can overwhelm target servers (causing 429 rate limit errors or even getting IP banned) or exhaust client resources (memory, file descriptors, network buffers). Launching 10,000 tasks simultaneously can cause OOM errors.

What’s the improvement: Rate limiting is respectful (doesn’t DDoS targets) and prevents resource exhaustion. Semaphore limits concurrent requests: instead of launching 10,000 tasks simultaneously, limit to 100 concurrent tasks. Token bucket rate limiter ensures you don’t exceed server rate limits (e.g., 100 req/min). This prevents bans and keeps your client healthy.

Optimization focus: Resource efficiency and reliability - stay within limits while maximizing throughput.

Architecture:

• Structs: RateLimiter, ResourceManager
• Functions:
  • RateLimiter::new(requests_per_second) - Create rate limiter
  • acquire() -> Permit - Wait for rate limit token
  • Use tokio::sync::Semaphore for concurrency limiting

Starter Code:

#![allow(unused)]
fn main() {
use tokio::sync::Semaphore;
use std::sync::Arc;

/// Token bucket rate limiter
/// Role: Enforce rate limits
pub struct RateLimiter {
    semaphore: Arc<Semaphore>,                      // Concurrency limiter              
    rate_per_second: f64,                           // Target rate                           
    last_request: Arc<Mutex<Instant>>,              // Last request timestamp   
}

impl RateLimiter {
    /// Create rate limiter
    /// Role: Initialize with rate limit
    pub fn new(max_concurrent: usize, rate_per_second: f64) -> Self {
        todo!("Create semaphore with max_concurrent permits")
    }

    /// Acquire permission to make request
    /// Role: Block until rate limit allows request
    pub async fn acquire(&self) -> tokio::sync::SemaphorePermit<'_> {
        todo!("Wait for semaphore, enforce rate limiting")
    }
}

/// Resource manager for scraper
/// Role: Monitor and limit resource consumption
pub struct ResourceManager {
    active_requests: Arc<Mutex<usize>>,                // Current in-flight requests   
    total_bytes: Arc<Mutex<u64>>,                      // Total data downloaded              
    max_memory_bytes: u64,                             // Memory limit                              
}

impl ResourceManager {
    /// Create resource manager
    /// Role: Initialize with limits
    pub fn new(max_memory_bytes: u64) -> Self {
        todo!("Initialize counters")
    }

    /// Check if can make new request
    /// Role: Enforce resource limits
    pub fn can_proceed(&self) -> bool {
        todo!("Check memory usage against limit")
    }

    /// Record request start
    /// Role: Track active request
    pub fn start_request(&self) {
        todo!("Increment active_requests")
    }

    /// Record request completion
    /// Role: Update counters
    pub fn end_request(&self, bytes: u64) {
        todo!("Decrement active_requests, add to total_bytes")
    }
}

/// Fetch URLs with complete resource management
/// Role: Production-ready scraping
pub async fn fetch_all_managed(
    urls: Vec<String>,
    timeout_ms: u64,
    retry_config: &RetryConfig,
    circuit_breaker: &Arc<CircuitBreaker>,
    rate_limiter: &RateLimiter,
    resource_manager: &Arc<ResourceManager>,
) -> Vec<FetchResult> {
    todo!("Combine rate limiting, circuit breaker, resource management")
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_rate_limiter_enforces_concurrency() {
        let limiter = RateLimiter::new(2, 100.0);

        let start = Instant::now();

        // Try to acquire 3 permits
        let _p1 = limiter.acquire().await;
        let _p2 = limiter.acquire().await;

        // Third should wait (spawn task to test non-blocking of test)
        let limiter_clone = Arc::new(limiter);
        let task = tokio::spawn({
            let limiter = limiter_clone.clone();
            async move {
                let _p3 = limiter.acquire().await;
            }
        });

        // Give it a moment
        tokio::time::sleep(Duration::from_millis(50)).await;

        drop(_p1); // Release one permit

        task.await.unwrap();
    }

    #[tokio::test]
    async fn test_rate_limiter_timing() {
        let limiter = RateLimiter::new(10, 5.0); // 5 requests/second

        let start = Instant::now();

        // Make 10 requests (should take ~2 seconds at 5/sec)
        for _ in 0..10 {
            let _permit = limiter.acquire().await;
        }

        let elapsed = start.elapsed();

        // Should take roughly 2 seconds
        assert!(elapsed.as_millis() >= 1800);
        assert!(elapsed.as_millis() < 2500);
    }

    #[tokio::test]
    async fn test_resource_manager_tracks_requests() {
        let manager = Arc::new(ResourceManager::new(1_000_000));

        manager.start_request();
        assert_eq!(*manager.active_requests.lock().unwrap(), 1);

        manager.start_request();
        assert_eq!(*manager.active_requests.lock().unwrap(), 2);

        manager.end_request(1000);
        assert_eq!(*manager.active_requests.lock().unwrap(), 1);
        assert_eq!(*manager.total_bytes.lock().unwrap(), 1000);

        manager.end_request(2000);
        assert_eq!(*manager.active_requests.lock().unwrap(), 0);
        assert_eq!(*manager.total_bytes.lock().unwrap(), 3000);
    }

    #[tokio::test]
    async fn test_resource_manager_enforces_limits() {
        let manager = Arc::new(ResourceManager::new(1000));

        manager.start_request();
        manager.end_request(500);

        assert!(manager.can_proceed());

        manager.start_request();
        manager.end_request(600);

        // Total is now 1100, exceeds limit
        assert!(!manager.can_proceed());
    }

    #[tokio::test]
    async fn test_fetch_all_managed_integration() {
        let mock_server = MockServer::start().await;

        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200).set_body_string("data"))
            .mount(&mock_server)
            .await;

        let urls: Vec<String> = (0..5)
            .map(|i| format!("{}/{}", mock_server.uri(), i))
            .collect();

        let config = RetryConfig::default();
        let cb = Arc::new(CircuitBreaker::new(5, Duration::from_secs(10)));
        let limiter = RateLimiter::new(2, 10.0);
        let manager = Arc::new(ResourceManager::new(1_000_000));

        let results = fetch_all_managed(
            urls,
            5000,
            &config,
            &cb,
            &limiter,
            &manager,
        ).await;

        assert_eq!(results.len(), 5);
        assert!(results.iter().all(|r| r.is_success()));
    }
}
}

Complete Working Example

#![allow(unused)]
fn main() {
use futures::future::join_all;
use rand::Rng;
use reqwest::Client;
use std::{
    future::Future,
    sync::{Arc, Mutex},
    time::{Duration, Instant},
};
use thiserror::Error;
use tokio::{
    sync::{OwnedSemaphorePermit, Semaphore},
    time::{sleep, timeout},
};

// =============================================================================
// Milestone 1: Basic Async HTTP Client with Error Types
// =============================================================================

#[derive(Error, Debug, Clone)]
pub enum ScraperError {
    #[error("Network error: {0}")]
    NetworkError(String),

    #[error("Request timed out after {0}ms")]
    TimeoutError(u64),

    #[error("HTTP {status} error for {url}")]
    HttpError { status: u16, url: String },

    #[error("Failed to parse response: {0}")]
    ParseError(String),

    #[error("Circuit breaker is open")]
    CircuitBreakerOpen,

    #[error("Resource limit exceeded: {0}")]
    ResourceLimitExceeded(String),
}

impl From<reqwest::Error> for ScraperError {
    fn from(err: reqwest::Error) -> Self {
        if err.is_timeout() {
            return ScraperError::TimeoutError(0);
        }

        if let Some(status) = err.status() {
            return ScraperError::HttpError {
                status: status.as_u16(),
                url: err.url().map(|u| u.to_string()).unwrap_or_default(),
            };
        }

        ScraperError::NetworkError(err.to_string())
    }
}

pub async fn fetch_url(url: &str) -> Result<String, ScraperError> {
    let client = Client::new();
    let response = client.get(url).send().await.map_err(ScraperError::from)?;

    if !response.status().is_success() {
        return Err(ScraperError::HttpError {
            status: response.status().as_u16(),
            url: url.to_string(),
        });
    }

    response
        .text()
        .await
        .map_err(|e| ScraperError::ParseError(e.to_string()))
}

// =============================================================================
// Milestone 2: Enforcing Timeouts on Network Operations
// =============================================================================

pub async fn fetch_url_with_timeout(url: &str, timeout_ms: u64) -> Result<String, ScraperError> {
    match timeout(Duration::from_millis(timeout_ms), fetch_url(url)).await {
        Ok(result) => result,
        Err(_) => Err(ScraperError::TimeoutError(timeout_ms)),
    }
}

pub fn create_client(timeout_ms: u64) -> Client {
    Client::builder()
        .timeout(Duration::from_millis(timeout_ms))
        .build()
        .expect("HTTP client")
}

// =============================================================================
// Milestone 3: Retry with Exponential Backoff + Jitter
// =============================================================================

#[derive(Debug, Clone)]
pub struct RetryConfig {
    pub max_attempts: usize,
    pub initial_backoff_ms: u64,
    pub max_backoff_ms: u64,
    pub jitter: bool,
}

impl RetryConfig {
    pub fn default() -> Self {
        Self {
            max_attempts: 3,
            initial_backoff_ms: 1_000,
            max_backoff_ms: 30_000,
            jitter: true,
        }
    }

    pub fn backoff_duration(&self, attempt: usize) -> Duration {
        let multiplier = 1u64
            .checked_shl(attempt as u32)
            .unwrap_or(u64::MAX);
        let base = self.initial_backoff_ms.saturating_mul(multiplier);
        let capped = base.min(self.max_backoff_ms);
        let millis = if self.jitter {
            let mut rng = rand::thread_rng();
            (capped as f64 * rng.gen_range(0.9..1.1)) as u64
        } else {
            capped
        };
        Duration::from_millis(millis)
    }
}

impl ScraperError {
    pub fn is_retryable(&self) -> bool {
        matches!(
            self,
            ScraperError::NetworkError(_)
                | ScraperError::TimeoutError(_)
                | ScraperError::HttpError { status: 500..=599, .. }
        )
    }
}

struct RetryOutcome {
    result: Result<String, ScraperError>,
    attempt_count: usize,
}

async fn fetch_with_retry_internal(
    client: &Client,
    url: &str,
    timeout_ms: u64,
    retry_config: &RetryConfig,
) -> RetryOutcome {
    let mut attempt = 0;
    let max_attempts = retry_config.max_attempts.max(1);

    loop {
        attempt += 1;
        let future = async {
            let response = client.get(url).send().await.map_err(ScraperError::from)?;
            if !response.status().is_success() {
                return Err(ScraperError::HttpError {
                    status: response.status().as_u16(),
                    url: url.to_string(),
                });
            }

            response
                .text()
                .await
                .map_err(|e| ScraperError::ParseError(e.to_string()))
        };
        let result = match timeout(Duration::from_millis(timeout_ms), future).await {
            Ok(output) => output,
            Err(_) => Err(ScraperError::TimeoutError(timeout_ms)),
        };

        match result {
            Ok(body) => {
                return RetryOutcome {
                    result: Ok(body),
                    attempt_count: attempt,
                }
            }
            Err(err) => {
                if attempt >= max_attempts || !err.is_retryable() {
                    return RetryOutcome {
                        result: Err(err),
                        attempt_count: attempt,
                    };
                }

                let delay = retry_config.backoff_duration(attempt - 1);
                sleep(delay).await;
            }
        }
    }
}

pub async fn fetch_with_retry(
    url: &str,
    timeout_ms: u64,
    retry_config: &RetryConfig,
) -> Result<String, ScraperError> {
    let client = create_client(timeout_ms);
    fetch_with_retry_internal(&client, url, timeout_ms, retry_config)
        .await
        .result
}

// =============================================================================
// Milestone 4: Circuit Breaker Pattern
// =============================================================================

#[derive(Debug, Clone, PartialEq)]
pub enum CircuitState {
    Closed,
    Open { opened_at: Instant },
    HalfOpen,
}

pub struct CircuitBreaker {
    state: Arc<Mutex<CircuitState>>,
    failure_threshold: usize,
    success_threshold: usize,
    timeout: Duration,
    consecutive_failures: Arc<Mutex<usize>>,
    consecutive_successes: Arc<Mutex<usize>>,
}

impl CircuitBreaker {
    pub fn new(failure_threshold: usize, timeout: Duration) -> Self {
        Self {
            state: Arc::new(Mutex::new(CircuitState::Closed)),
            failure_threshold: failure_threshold.max(1),
            success_threshold: 1,
            timeout,
            consecutive_failures: Arc::new(Mutex::new(0)),
            consecutive_successes: Arc::new(Mutex::new(0)),
        }
    }

    pub fn state(&self) -> CircuitState {
        self.state.lock().unwrap().clone()
    }

    fn should_attempt(&self) -> bool {
        let mut state = self.state.lock().unwrap();
        match *state {
            CircuitState::Open { opened_at } => {
                if opened_at.elapsed() >= self.timeout {
                    *state = CircuitState::HalfOpen;
                    true
                } else {
                    false
                }
            }
            _ => true,
        }
    }

    pub async fn call<F, T>(&self, f: F) -> Result<T, ScraperError>
    where
        F: Future<Output = Result<T, ScraperError>>,
    {
        if !self.should_attempt() {
            return Err(ScraperError::CircuitBreakerOpen);
        }

        match f.await {
            Ok(value) => {
                self.on_success();
                Ok(value)
            }
            Err(err) => {
                self.on_failure();
                Err(err)
            }
        }
    }

    fn on_success(&self) {
        *self.consecutive_failures.lock().unwrap() = 0;
        let mut successes = self.consecutive_successes.lock().unwrap();
        *successes += 1;

        let mut state = self.state.lock().unwrap();
        if matches!(*state, CircuitState::HalfOpen) && *successes >= self.success_threshold {
            *state = CircuitState::Closed;
            *successes = 0;
        }
    }

    fn on_failure(&self) {
        *self.consecutive_successes.lock().unwrap() = 0;
        let mut failures = self.consecutive_failures.lock().unwrap();
        *failures += 1;

        if *failures >= self.failure_threshold {
            let mut state = self.state.lock().unwrap();
            *state = CircuitState::Open {
                opened_at: Instant::now(),
            };
            *failures = 0;
        }
    }
}

// =============================================================================
// Milestone 5: Concurrent Fetching + Partial Results
// =============================================================================

#[derive(Debug)]
pub struct FetchResult {
    pub url: String,
    pub result: Result<String, ScraperError>,
    pub duration_ms: u64,
    pub attempt_count: usize,
}

impl FetchResult {
    pub fn is_success(&self) -> bool {
        self.result.is_ok()
    }

    pub fn content(&self) -> Option<&str> {
        self.result.as_ref().ok().map(|s| s.as_str())
    }
}

#[derive(Debug)]
pub struct FetchSummary {
    pub total: usize,
    pub success: usize,
    pub failed: usize,
    pub total_duration_ms: u64,
    pub avg_duration_ms: u64,
}

impl FetchSummary {
    pub fn from_results(results: &[FetchResult]) -> Self {
        let total = results.len();
        let success = results.iter().filter(|r| r.is_success()).count();
        let failed = total.saturating_sub(success);
        let total_duration_ms: u64 = results.iter().map(|r| r.duration_ms).sum();
        let avg_duration_ms = if total > 0 {
            total_duration_ms / total as u64
        } else {
            0
        };

        Self {
            total,
            success,
            failed,
            total_duration_ms,
            avg_duration_ms,
        }
    }

    pub fn success_rate(&self) -> f64 {
        if self.total == 0 {
            0.0
        } else {
            (self.success as f64 / self.total as f64) * 100.0
        }
    }
}

pub async fn fetch_all(
    urls: Vec<String>,
    timeout_ms: u64,
    retry_config: &RetryConfig,
    circuit_breaker: &Arc<CircuitBreaker>,
) -> Vec<FetchResult> {
    let client = create_client(timeout_ms);

    let futures = urls.into_iter().map(|url| {
        let client = client.clone();
        let retry = retry_config.clone();
        let breaker = circuit_breaker.clone();

        async move {
            let start = Instant::now();
            let attempts = Arc::new(Mutex::new(0_usize));
            let attempts_inner = attempts.clone();

            let result = breaker
                .call(async {
                    let outcome = fetch_with_retry_internal(&client, &url, timeout_ms, &retry).await;
                    *attempts_inner.lock().unwrap() = outcome.attempt_count;
                    outcome.result
                })
                .await;

            let attempt_count = {
                let guard = attempts.lock().unwrap();
                *guard
            };

            FetchResult {
                url,
                result,
                duration_ms: start.elapsed().as_millis() as u64,
                attempt_count,
            }
        }
    });

    join_all(futures).await
}

pub async fn fetch_all_with_limit(
    urls: Vec<String>,
    timeout_ms: u64,
    retry_config: &RetryConfig,
    circuit_breaker: &Arc<CircuitBreaker>,
    max_concurrent: usize,
) -> Vec<FetchResult> {
    let semaphore = Arc::new(Semaphore::new(max_concurrent.max(1)));
    let client = create_client(timeout_ms);

    let futures = urls.into_iter().map(|url| {
        let semaphore = semaphore.clone();
        let client = client.clone();
        let retry = retry_config.clone();
        let breaker = circuit_breaker.clone();

        async move {
            let _permit = semaphore.acquire_owned().await.expect("permit");
            let start = Instant::now();
            let attempts = Arc::new(Mutex::new(0_usize));
            let attempts_inner = attempts.clone();

            let result = breaker
                .call(async {
                    let outcome = fetch_with_retry_internal(&client, &url, timeout_ms, &retry).await;
                    *attempts_inner.lock().unwrap() = outcome.attempt_count;
                    outcome.result
                })
                .await;

            let attempt_count = {
                let guard = attempts.lock().unwrap();
                *guard
            };

            FetchResult {
                url,
                result,
                duration_ms: start.elapsed().as_millis() as u64,
                attempt_count,
            }
        }
    });

    join_all(futures).await
}

// =============================================================================
// Milestone 6: Rate Limiting and Resource Management
// =============================================================================

#[derive(Clone)]
pub struct RateLimiter {
    semaphore: Arc<Semaphore>,
    rate_per_second: f64,
    last_request: Arc<Mutex<Instant>>,
}

impl RateLimiter {
    pub fn new(max_concurrent: usize, rate_per_second: f64) -> Self {
        Self {
            semaphore: Arc::new(Semaphore::new(max_concurrent.max(1))),
            rate_per_second: rate_per_second.max(0.1),
            last_request: Arc::new(Mutex::new(Instant::now() - Duration::from_secs(1))),
        }
    }

    pub async fn acquire(&self) -> OwnedSemaphorePermit {
        let permit = self.semaphore.clone().acquire_owned().await.expect("permit");
        let interval = Duration::from_secs_f64(1.0 / self.rate_per_second);
        let mut last = self.last_request.lock().unwrap();
        let now = Instant::now();
        let next_allowed = *last + interval;
        if now < next_allowed {
            sleep(next_allowed - now).await;
        }
        *last = Instant::now();
        permit
    }
}

pub struct ResourceManager {
    active_requests: Arc<Mutex<usize>>,
    total_bytes: Arc<Mutex<u64>>,
    max_memory_bytes: u64,
}

impl ResourceManager {
    pub fn new(max_memory_bytes: u64) -> Self {
        Self {
            active_requests: Arc::new(Mutex::new(0)),
            total_bytes: Arc::new(Mutex::new(0)),
            max_memory_bytes,
        }
    }

    pub fn can_proceed(&self) -> bool {
        *self.total_bytes.lock().unwrap() <= self.max_memory_bytes
    }

    pub fn start_request(&self) {
        *self.active_requests.lock().unwrap() += 1;
    }

    pub fn end_request(&self, bytes: u64) {
        let mut active = self.active_requests.lock().unwrap();
        if *active > 0 {
            *active -= 1;
        }
        *self.total_bytes.lock().unwrap() += bytes;
    }
}

pub async fn fetch_all_managed(
    urls: Vec<String>,
    timeout_ms: u64,
    retry_config: &RetryConfig,
    circuit_breaker: &Arc<CircuitBreaker>,
    rate_limiter: &RateLimiter,
    resource_manager: &Arc<ResourceManager>,
) -> Vec<FetchResult> {
    let client = create_client(timeout_ms);

    let futures = urls.into_iter().map(|url| {
        let client = client.clone();
        let retry = retry_config.clone();
        let breaker = circuit_breaker.clone();
        let limiter = rate_limiter.clone();
        let manager = resource_manager.clone();

        async move {
            let permit = limiter.acquire().await;
            if !manager.can_proceed() {
                drop(permit);
                return FetchResult {
                    url,
                    result: Err(ScraperError::ResourceLimitExceeded(
                        "memory limit reached".to_string(),
                    )),
                    duration_ms: 0,
                    attempt_count: 0,
                };
            }

            manager.start_request();
            let start = Instant::now();
            let attempts = Arc::new(Mutex::new(0_usize));
            let attempts_inner = attempts.clone();

            let result = breaker
                .call(async {
                    let outcome = fetch_with_retry_internal(&client, &url, timeout_ms, &retry).await;
                    *attempts_inner.lock().unwrap() = outcome.attempt_count;
                    outcome.result
                })
                .await;

            let attempt_count = {
                let guard = attempts.lock().unwrap();
                *guard
            };

            let bytes = result.as_ref().ok().map(|body| body.len() as u64).unwrap_or(0);
            manager.end_request(bytes);
            drop(permit);

            FetchResult {
                url,
                result,
                duration_ms: start.elapsed().as_millis() as u64,
                attempt_count,
            }
        }
    });

    join_all(futures).await
}

// =============================================================================
// Tests for All Milestones
// =============================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tokio::time::Duration;
    use wiremock::{matchers::{method, path}, Mock, MockServer, ResponseTemplate};

    #[tokio::test]
    async fn test_fetch_url_success() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/test"))
            .respond_with(ResponseTemplate::new(200).set_body_string("Hello"))
            .mount(&server)
            .await;

        let url = format!("{}/test", server.uri());
        let result = fetch_url(&url).await.unwrap();
        assert_eq!(result, "Hello");
    }

    #[tokio::test]
    async fn test_fetch_url_404() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(404))
            .mount(&server)
            .await;

        let err = fetch_url(&server.uri()).await.unwrap_err();
        match err {
            ScraperError::HttpError { status, .. } => assert_eq!(status, 404),
            _ => panic!("expected HTTP error"),
        }
    }

    #[tokio::test]
    async fn test_fetch_url_with_timeout_triggers() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200).set_delay(Duration::from_secs(2)))
            .mount(&server)
            .await;

        let err = fetch_url_with_timeout(&server.uri(), 50).await.unwrap_err();
        match err {
            ScraperError::TimeoutError(ms) => assert_eq!(ms, 50),
            _ => panic!("expected timeout"),
        }
    }

    #[test]
    fn test_retry_config_backoff() {
        let cfg = RetryConfig {
            max_attempts: 5,
            initial_backoff_ms: 100,
            max_backoff_ms: 500,
            jitter: false,
        };
        assert_eq!(cfg.backoff_duration(0).as_millis(), 100);
        assert_eq!(cfg.backoff_duration(1).as_millis(), 200);
        assert_eq!(cfg.backoff_duration(2).as_millis(), 400);
        assert_eq!(cfg.backoff_duration(3).as_millis(), 500);
    }

    #[tokio::test]
    async fn test_fetch_with_retry_succeeds_after_failures() {
        let server = MockServer::start().await;
        let attempts = Arc::new(AtomicUsize::new(0));
        let attempts_clone = attempts.clone();

        Mock::given(method("GET"))
            .respond_with(move |_req: &wiremock::Request| {
                let count = attempts_clone.fetch_add(1, Ordering::SeqCst);
                if count < 2 {
                    ResponseTemplate::new(500)
                } else {
                    ResponseTemplate::new(200).set_body_string("ok")
                }
            })
            .mount(&server)
            .await;

        let cfg = RetryConfig {
            max_attempts: 3,
            initial_backoff_ms: 10,
            max_backoff_ms: 100,
            jitter: false,
        };

        let result = fetch_with_retry(&server.uri(), 1000, &cfg).await.unwrap();
        assert_eq!(result, "ok");
        assert_eq!(attempts.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn test_circuit_breaker_flow() {
        let breaker = Arc::new(CircuitBreaker::new(2, Duration::from_millis(100)));

        for _ in 0..2 {
            let _ = breaker
                .call(async { Err::<(), _>(ScraperError::NetworkError("fail".into())) })
                .await;
        }

        match breaker.state() {
            CircuitState::Open { .. } => {}
            _ => panic!("expected open"),
        }

        tokio::time::sleep(Duration::from_millis(120)).await;
        let res = breaker
            .call(async { Ok::<_, ScraperError>("ok") })
            .await
            .unwrap();
        assert_eq!(res, "ok");
        assert_eq!(breaker.state(), CircuitState::Closed);
    }

    #[tokio::test]
    async fn test_fetch_all_partial_results() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/good"))
            .respond_with(ResponseTemplate::new(200).set_body_string("good"))
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/bad"))
            .respond_with(ResponseTemplate::new(500))
            .mount(&server)
            .await;

        let urls = vec![
            format!("{}/good", server.uri()),
            format!("{}/bad", server.uri()),
        ];
        let cfg = RetryConfig {
            max_attempts: 1,
            ..RetryConfig::default()
        };
        let breaker = Arc::new(CircuitBreaker::new(5, Duration::from_secs(1)));

        let results = fetch_all(urls, 1_000, &cfg, &breaker).await;
        assert_eq!(results.len(), 2);
        assert!(results[0].is_success());
        assert!(!results[1].is_success());
    }

    #[tokio::test]
    async fn test_fetch_all_with_limit_respects_limit() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .respond_with(
                ResponseTemplate::new(200).set_delay(Duration::from_millis(100)),
            )
            .mount(&server)
            .await;

        let urls: Vec<_> = (0..6).map(|i| format!("{}/{}", server.uri(), i)).collect();
        let cfg = RetryConfig::default();
        let breaker = Arc::new(CircuitBreaker::new(10, Duration::from_secs(1)));

        let start = Instant::now();
        let _results = fetch_all_with_limit(urls, 1_000, &cfg, &breaker, 2).await;
        assert!(start.elapsed().as_millis() >= 300);
    }

    #[test]
    fn test_fetch_summary() {
        let results = vec![
            FetchResult {
                url: "a".into(),
                result: Ok("1".into()),
                duration_ms: 100,
                attempt_count: 1,
            },
            FetchResult {
                url: "b".into(),
                result: Err(ScraperError::NetworkError("fail".into())),
                duration_ms: 200,
                attempt_count: 2,
            },
        ];

        let summary = FetchSummary::from_results(&results);
        assert_eq!(summary.total, 2);
        assert_eq!(summary.success, 1);
        assert_eq!(summary.failed, 1);
        assert_eq!(summary.avg_duration_ms, 150);
        assert!((summary.success_rate() - 50.0).abs() < f64::EPSILON);
    }

    #[tokio::test]
    async fn test_rate_limiter_enforces_rate() {
        let limiter = RateLimiter::new(2, 5.0);
        let start = Instant::now();

        for _ in 0..5 {
            let _permit = limiter.acquire().await;
        }

        assert!(start.elapsed().as_millis() >= 800);
    }

    #[tokio::test]
    async fn test_resource_manager_tracks_usage() {
        let manager = Arc::new(ResourceManager::new(1_000));
        assert!(manager.can_proceed());
        manager.start_request();
        manager.end_request(500);
        assert!(manager.can_proceed());
        manager.start_request();
        manager.end_request(600);
        assert!(!manager.can_proceed());
    }

    #[tokio::test]
    async fn test_fetch_all_managed() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200).set_body_string("data"))
            .mount(&server)
            .await;

        let urls: Vec<_> = (0..3).map(|i| format!("{}/{}", server.uri(), i)).collect();
        let cfg = RetryConfig::default();
        let breaker = Arc::new(CircuitBreaker::new(5, Duration::from_secs(1)));
        let limiter = RateLimiter::new(2, 10.0);
        let manager = Arc::new(ResourceManager::new(10_000));

        let results = fetch_all_managed(urls, 1_000, &cfg, &breaker, &limiter, &manager).await;
        assert_eq!(results.len(), 3);
        assert!(results.iter().all(|r| r.is_success()));
    }
}

}

Project-Wide Benefits

Resilience patterns combined:

Measured improvements (scraping 1000 URLs):

When to use these patterns:

• ✅ Web scraping: All patterns essential
• ✅ API aggregation: Multiple external APIs
• ✅ Microservices: Service-to-service calls
• ✅ Health monitoring: Check distributed systems
• ❌ Single server: Retry/circuit breaker overkill
• ❌ Trusted network: Timeout less critical