Implementing Adaptive Backpressure in Rust with FlowGuard

Source: Dev.to

The Problem with Static Limits

// "Maximum 100 concurrent connections"
let max_connections = 100;

Static limits are a trap:

• Set too high? Your system crashes before reaching the limit.
• Set too low? You waste resources and refuse legitimate traffic.
• Guessing game? You’re always tuning based on hunches.

The Solution: Dynamic Backpressure

Instead of guessing, let your system self‑adjust based on real‑time performance. That’s where FlowGuard comes in.

Introducing FlowGuard

FlowGuard implements the TCP Vegas congestion control algorithm to dynamically adjust concurrency limits based on actual system latency.

How It Works

use flow_guard::{FlowGuard, VegasStrategy};
use std::sync::Arc;

#[tokio::main]
async fn main() {
    // Start with 10 concurrent operations
    let strategy = Arc::new(VegasStrategy::new(10));
    let guard = FlowGuard::new(Arc::clone(&strategy));

    println!("Initial limit: {}", guard.current_limit());

    // Execute tasks with adaptive backpressure
    let result = guard
        .run(async {
            // Your database query, API call, etc.
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            Ok::("Success!")
        })
        .await;

    println!("Final limit: {}", guard.current_limit()); // Adjusted!
}

Key Features

Real‑time Adjustment

// Watch limits adjust dynamically
println!("Current limit: {}", guard.current_limit());
println!("Available permits: {}", guard.available_permits());

Vegas Algorithm

• Increases limit when the system has spare capacity.
• Decreases limit when latency indicates congestion.
• Self‑tuning – no manual configuration needed.

Web Framework Integration

// Axum 0.8 middleware
let strategy = VegasStrategy::new(50);
let flow_layer = FlowGuardLayer::new(strategy);

let app = Router::new()
    .route("/api/data", get(handler))
    .layer(flow_layer);

Getting Started

Add to your Cargo.toml:

[dependencies]
flow-guard = "0.2.1"

With Axum/Tower support:

flow-guard = { version = "0.2.1", features = ["axum", "tower"] }

Under the Hood

FlowGuard replaces tokio::sync::Semaphore with a custom DynamicSemaphore that can adjust its limit up and down in real time:

pub struct DynamicSemaphore {
    max_permits: AtomicUsize,
    available_permits: AtomicUsize,
    notify: Notify,
}

impl DynamicSemaphore {
    pub fn set_limit(&self, new_limit: usize) {
        // Adjusts permits dynamically based on Vegas calculations
    }
}

Use Cases

Database Protection

// Prevent database overload
let db_guard = FlowGuard::new(VegasStrategy::new(20));

async fn query_database() -> Result> {
    db_guard
        .run(|| async {
            // Your database query here
            database.query("SELECT * FROM users").await
        })
        .await
}

API Rate Limiting

// Adaptive rate limiting for external APIs
let api_guard = FlowGuard::new(VegasStrategy::new(5));

async fn call_external_api() -> Result> {
    api_guard
        .run(|| async {
            client
                .get("https://api.example.com/data")
                .await?
                .text()
                .await
        })
        .await
}

Microservices

// Protect services from cascading failures
let service_guard = FlowGuard::new(VegasStrategy::new(100));

Benchmarks

In testing, FlowGuard showed:

• 5 → 12 limit adjustment under optimal conditions
• Sub‑millisecond overhead per request
• Zero allocation in hot path
• Thread‑safe with atomic operations

Try It Yourself

# Clone and run examples
git clone https://github.com/cleitonaugusto/flow-guard
cd flow-guard
cargo run --example basic_usage

Resources

• GitHub:
• Crates.io:
• Documentation:
• Examples: basic_usage.rs, server_demo.rs

Why I Built This

Static limits either crash services or waste resources. The TCP Vegas algorithm has been battle‑tested for decades in networking—why not apply it to service concurrency?

Contributing & Feedback

FlowGuard is open source under the MIT license. Contributions are welcome:

• Feedback on the API design
• Use cases from your projects
• Code contributions and improvements

What adaptive concurrency patterns have you used in your Rust projects? Share your thoughts in the comments!