Struct Bulkhead 

pub struct Bulkhead { /* private fields */ }

A policy that limits concurrent executions.

The bulkhead uses a semaphore to control how many operations can run simultaneously. When the limit is reached, new requests are either rejected immediately or queued (if a queue timeout is configured).

Examples

use do_over::{policy::Policy, bulkhead::Bulkhead, error::DoOverError};
use std::time::Duration;

// Basic bulkhead - reject immediately when full
let bulkhead = Bulkhead::new(5);

// With queue timeout - wait for a slot
let bulkhead = Bulkhead::new(5)
    .with_queue_timeout(Duration::from_millis(500));

Implementations

impl Bulkhead

pub fn new(max_concurrent: usize) -> Self

Create a new bulkhead with the specified concurrency limit.

Without a queue timeout, requests are rejected immediately when no slots are available.

Arguments

• max_concurrent - Maximum number of concurrent executions

Examples

use do_over::bulkhead::Bulkhead;

// Allow up to 10 concurrent operations
let bulkhead = Bulkhead::new(10);

Examples found in repository

examples/bulkhead.rs (line 29)

async fn basic_concurrency_example() {
    println!("📌 Scenario 1: Basic Concurrency Limiting");
    println!("   Configuration: max_concurrent=2, no queue");
    println!("   Launching 5 concurrent requests...\n");

    let bulkhead = Arc::new(Bulkhead::new(2));
    let start = Instant::now();
    let completed = Arc::new(AtomicUsize::new(0));
    let rejected = Arc::new(AtomicUsize::new(0));

    let mut handles = vec![];

    for i in 1..=5 {
        let bh = Arc::clone(&bulkhead);
        let comp = Arc::clone(&completed);
        let rej = Arc::clone(&rejected);
        let s = start;

        let handle = tokio::spawn(async move {
            let result: Result<String, DoOverError<String>> = bh
                .execute(|| async {
                    let elapsed = s.elapsed().as_millis();
                    println!(
                        "   [+{:>4}ms] Request {}: 🔓 Acquired slot, processing...",
                        elapsed, i
                    );
                    tokio::time::sleep(Duration::from_millis(200)).await;
                    let elapsed = s.elapsed().as_millis();
                    println!(
                        "   [+{:>4}ms] Request {}: ✅ Completed, releasing slot",
                        elapsed, i
                    );
                    Ok(format!("Request {} done", i))
                })
                .await;

            match result {
                Ok(_) => {
                    comp.fetch_add(1, Ordering::SeqCst);
                }
                Err(DoOverError::BulkheadFull) => {
                    let elapsed = s.elapsed().as_millis();
                    println!(
                        "   [+{:>4}ms] Request {}: 🚫 Rejected (BulkheadFull)",
                        elapsed, i
                    );
                    rej.fetch_add(1, Ordering::SeqCst);
                }
                Err(e) => println!("   Request {}: Error - {:?}", i, e),
            }
        });
        handles.push(handle);

        // Small delay to make output more readable
        tokio::time::sleep(Duration::from_millis(10)).await;
    }

    for handle in handles {
        handle.await.unwrap();
    }

    let elapsed = start.elapsed().as_millis();
    println!("\n   Summary:");
    println!("   - Completed: {}", completed.load(Ordering::SeqCst));
    println!("   - Rejected:  {}", rejected.load(Ordering::SeqCst));
    println!("   - Total time: {}ms", elapsed);
    println!("\n   💡 Only 2 requests could run concurrently, 3 were rejected immediately");
}

async fn queue_timeout_example() {
    println!("📌 Scenario 2: With Queue Timeout");
    println!("   Configuration: max_concurrent=2, queue_timeout=150ms");
    println!("   Launching 5 concurrent requests...\n");

    let bulkhead = Arc::new(Bulkhead::new(2).with_queue_timeout(Duration::from_millis(150)));
    let start = Instant::now();
    let completed = Arc::new(AtomicUsize::new(0));
    let rejected = Arc::new(AtomicUsize::new(0));
    let queued = Arc::new(AtomicUsize::new(0));

    let mut handles = vec![];

    for i in 1..=5 {
        let bh = Arc::clone(&bulkhead);
        let comp = Arc::clone(&completed);
        let rej = Arc::clone(&rejected);
        let q = Arc::clone(&queued);
        let s = start;

        let handle = tokio::spawn(async move {
            let request_start = s.elapsed().as_millis();
            let result: Result<String, DoOverError<String>> = bh
                .execute(|| async {
                    let elapsed = s.elapsed().as_millis();
                    let wait_time = elapsed - request_start;
                    if wait_time > 10 {
                        println!(
                            "   [+{:>4}ms] Request {}: ⏳ Waited {}ms in queue, now processing",
                            elapsed, i, wait_time
                        );
                        q.fetch_add(1, Ordering::SeqCst);
                    } else {
                        println!(
                            "   [+{:>4}ms] Request {}: 🔓 Acquired slot immediately",
                            elapsed, i
                        );
                    }
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    let elapsed = s.elapsed().as_millis();
                    println!(
                        "   [+{:>4}ms] Request {}: ✅ Completed",
                        elapsed, i
                    );
                    Ok(format!("Request {} done", i))
                })
                .await;

            match result {
                Ok(_) => {
                    comp.fetch_add(1, Ordering::SeqCst);
                }
                Err(DoOverError::BulkheadFull) => {
                    let elapsed = s.elapsed().as_millis();
                    println!(
                        "   [+{:>4}ms] Request {}: 🚫 Queue timeout expired",
                        elapsed, i
                    );
                    rej.fetch_add(1, Ordering::SeqCst);
                }
                Err(e) => println!("   Request {}: Error - {:?}", i, e),
            }
        });
        handles.push(handle);

        // Small delay between launching requests
        tokio::time::sleep(Duration::from_millis(20)).await;
    }

    for handle in handles {
        handle.await.unwrap();
    }

    let elapsed = start.elapsed().as_millis();
    println!("\n   Summary:");
    println!("   - Completed:     {}", completed.load(Ordering::SeqCst));
    println!("   - Queued/Waited: {}", queued.load(Ordering::SeqCst));
    println!("   - Rejected:      {}", rejected.load(Ordering::SeqCst));
    println!("   - Total time:    {}ms", elapsed);
    println!("\n   💡 Requests waited in queue up to 150ms before being rejected");
}

examples/composition.rs (line 160)

async fn pattern3_full_stack() {
    println!("📌 Pattern 3: Full Resilience Stack");
    println!("   Bulkhead(2) → Retry(2, 100ms) → Timeout(500ms)");
    println!("   Recommended ordering for production systems\n");

    // Note: Using Bulkhead + Retry + Timeout for this example to avoid
    // CircuitBreaker clone issues in concurrent async context
    let policy = Arc::new(Wrap::new(
        Bulkhead::new(2),
        Wrap::new(
            RetryPolicy::fixed(2, Duration::from_millis(100)),
            TimeoutPolicy::new(Duration::from_millis(500)),
        ),
    ));

    let call_count = Arc::new(AtomicUsize::new(0));
    let start = Instant::now();

    println!("   Launching 4 concurrent requests (bulkhead limit is 2)...\n");

    let mut handles = vec![];
    for i in 1..=4 {
        let p = Arc::clone(&policy);
        let cc = Arc::clone(&call_count);
        let s = start;

        let handle = tokio::spawn(async move {
            let result: Result<String, DoOverError<String>> = p
                .execute(|| {
                    let count = Arc::clone(&cc);
                    async move {
                        let call = count.fetch_add(1, Ordering::SeqCst) + 1;
                        let elapsed = s.elapsed().as_millis();
                        println!(
                            "   [+{:>4}ms] Request {}, call {}: Processing...",
                            elapsed, i, call
                        );
                        tokio::time::sleep(Duration::from_millis(200)).await;
                        let elapsed = s.elapsed().as_millis();
                        println!(
                            "   [+{:>4}ms] Request {}, call {}: ✅ Done",
                            elapsed, i, call
                        );
                        Ok(format!("Response {}", i))
                    }
                })
                .await;

            let elapsed = s.elapsed().as_millis();
            match &result {
                Ok(msg) => println!("   [+{:>4}ms] Request {}: ✅ {}", elapsed, i, msg),
                Err(DoOverError::BulkheadFull) => {
                    println!("   [+{:>4}ms] Request {}: 🚫 Rejected by bulkhead", elapsed, i)
                }
                Err(e) => println!("   [+{:>4}ms] Request {}: ❌ {:?}", elapsed, i, e),
            }
        });
        handles.push(handle);

        tokio::time::sleep(Duration::from_millis(10)).await;
    }

    for handle in handles {
        handle.await.unwrap();
    }

    let total_time = start.elapsed().as_millis();
    let total_calls = call_count.load(Ordering::SeqCst);

    println!("\n   Summary:");
    println!("   - Total inner operation calls: {}", total_calls);
    println!("   - Total time: {}ms", total_time);
    println!("\n   Policy execution order:");
    println!("   1. Bulkhead: Limits to 2 concurrent executions");
    println!("   2. Retry: Retries transient failures");
    println!("   3. Timeout: Each attempt bounded to 500ms");
}

pub fn with_queue_timeout(self, timeout: Duration) -> Self

Set a queue timeout for waiting on a slot.

When set, requests will wait up to the specified duration for a slot to become available before being rejected.

Arguments

• timeout - Maximum time to wait for a slot

Examples

use do_over::bulkhead::Bulkhead;
use std::time::Duration;

let bulkhead = Bulkhead::new(10)
    .with_queue_timeout(Duration::from_secs(1));

Examples found in repository

examples/bulkhead.rs (line 98)

async fn queue_timeout_example() {
    println!("📌 Scenario 2: With Queue Timeout");
    println!("   Configuration: max_concurrent=2, queue_timeout=150ms");
    println!("   Launching 5 concurrent requests...\n");

    let bulkhead = Arc::new(Bulkhead::new(2).with_queue_timeout(Duration::from_millis(150)));
    let start = Instant::now();
    let completed = Arc::new(AtomicUsize::new(0));
    let rejected = Arc::new(AtomicUsize::new(0));
    let queued = Arc::new(AtomicUsize::new(0));

    let mut handles = vec![];

    for i in 1..=5 {
        let bh = Arc::clone(&bulkhead);
        let comp = Arc::clone(&completed);
        let rej = Arc::clone(&rejected);
        let q = Arc::clone(&queued);
        let s = start;

        let handle = tokio::spawn(async move {
            let request_start = s.elapsed().as_millis();
            let result: Result<String, DoOverError<String>> = bh
                .execute(|| async {
                    let elapsed = s.elapsed().as_millis();
                    let wait_time = elapsed - request_start;
                    if wait_time > 10 {
                        println!(
                            "   [+{:>4}ms] Request {}: ⏳ Waited {}ms in queue, now processing",
                            elapsed, i, wait_time
                        );
                        q.fetch_add(1, Ordering::SeqCst);
                    } else {
                        println!(
                            "   [+{:>4}ms] Request {}: 🔓 Acquired slot immediately",
                            elapsed, i
                        );
                    }
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    let elapsed = s.elapsed().as_millis();
                    println!(
                        "   [+{:>4}ms] Request {}: ✅ Completed",
                        elapsed, i
                    );
                    Ok(format!("Request {} done", i))
                })
                .await;

            match result {
                Ok(_) => {
                    comp.fetch_add(1, Ordering::SeqCst);
                }
                Err(DoOverError::BulkheadFull) => {
                    let elapsed = s.elapsed().as_millis();
                    println!(
                        "   [+{:>4}ms] Request {}: 🚫 Queue timeout expired",
                        elapsed, i
                    );
                    rej.fetch_add(1, Ordering::SeqCst);
                }
                Err(e) => println!("   Request {}: Error - {:?}", i, e),
            }
        });
        handles.push(handle);

        // Small delay between launching requests
        tokio::time::sleep(Duration::from_millis(20)).await;
    }

    for handle in handles {
        handle.await.unwrap();
    }

    let elapsed = start.elapsed().as_millis();
    println!("\n   Summary:");
    println!("   - Completed:     {}", completed.load(Ordering::SeqCst));
    println!("   - Queued/Waited: {}", queued.load(Ordering::SeqCst));
    println!("   - Rejected:      {}", rejected.load(Ordering::SeqCst));
    println!("   - Total time:    {}ms", elapsed);
    println!("\n   💡 Requests waited in queue up to 150ms before being rejected");
}

Trait Implementations

impl Clone for Bulkhead

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<E> Policy<DoOverError<E>> for Bulkhead

where E: Send + Sync,

fn execute<'life0, 'async_trait, F, Fut, T>( &'life0 self, f: F, ) -> Pin<Box<dyn Future<Output = Result<T, DoOverError<E>>> + Send + 'async_trait>>

where F: Fn() -> Fut + Send + Sync + 'async_trait, Fut: Future<Output = Result<T, DoOverError<E>>> + Send + 'async_trait, T: Send + 'async_trait, Self: 'async_trait, 'life0: 'async_trait,

Execute an async operation with this policy’s resilience behavior.