use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;
use tokio::time::sleep;
#[derive(Debug)]
struct BucketState {
tokens: f64,
last_refill: Instant,
}
#[derive(Clone)]
pub struct TokenBucket {
capacity: f64,
refill_per_sec: f64,
state: Arc<Mutex<BucketState>>,
}
impl TokenBucket {
pub fn new(capacity: f64, refill_per_sec: f64) -> Self {
let state = BucketState {
tokens: capacity,
last_refill: Instant::now(),
};
TokenBucket {
capacity,
refill_per_sec,
state: Arc::new(Mutex::new(state)),
}
}
pub async fn take(&self, amount: f64) -> () {
loop {
let wait = {
let mut st = self.state.lock().await;
let now = Instant::now();
let elapsed = now.duration_since(st.last_refill).as_secs_f64();
st.tokens = (st.tokens + elapsed * self.refill_per_sec).min(self.capacity);
st.last_refill = now;
if st.tokens >= amount {
st.tokens -= amount;
return;
}
let deficit = amount - st.tokens;
Duration::from_secs_f64(deficit / self.refill_per_sec)
}; // guard dropped here, before the await
sleep(wait).await;
}
}
}
use reqwest::Client;
use crate::token_bucket::TokenBucket;
#[derive(Clone)]
pub struct RateLimitedClient {
inner: Client,
limiter: TokenBucket,
}
impl RateLimitedClient {
pub fn new(rps: f64, burst: f64) -> Self {
RateLimitedClient {
inner: Client::new(),
limiter: TokenBucket::new(burst, rps),
}
}
pub async fn get(&self, url: &str) -> reqwest::Result<String> {
self.limiter.take(1.0).await;
let resp = self.inner.get(url).send().await?;
resp.error_for_status()?.text().await
}
}
mod token_bucket;
mod client;
use std::time::Instant;
use client::RateLimitedClient;
#[tokio::main]
async fn main() {
let client = RateLimitedClient::new(5.0, 5.0); // 5 rps, burst of 5
let start = Instant::now();
let mut handles = Vec::new();
for id in 0..20 {
let client = client.clone();
handles.push(tokio::spawn(async move {
let url = format!("https://httpbin.org/anything?req={}", id);
match client.get(&url).await {
Ok(_) => println!("req {:>2} done at {:?}", id, start.elapsed()),
Err(e) => eprintln!("req {:>2} failed: {}", id, e),
}
}));
}
for h in handles {
let _ = h.await;
}
}
This snippet shows how to throttle outbound HTTP requests in an async Rust service using the classic token bucket algorithm. A token bucket refills at a steady rate up to a fixed capacity; each request must acquire a token before proceeding, and callers wait when the bucket is empty. This smooths bursts (up to capacity) while enforcing a long-run average rate, which is exactly what upstream APIs with quotas expect.
The TokenBucket tab implements the core. Rather than spawning a background refill task, it computes tokens lazily in take(): on each call it measures elapsed time since last_refill and adds elapsed * refill_per_sec, capped at capacity. This lazy approach avoids a timer thread and keeps the bucket exact even under irregular polling. The state lives behind a tokio::sync::Mutex wrapped in an Arc so many tasks can share one limiter. When fewer than one token is available, take() calculates the precise Duration until the next token accrues and sleeps for it, then loops — meaning take() always returns once a token is granted, providing natural backpressure.
A subtle detail: the mutex is dropped before sleeping. Holding a lock across an .await would serialize every waiter and defeat concurrency, so the code copies the needed values, releases the guard, and only then awaits. The f64 accumulation of fractional tokens lets sub-request-per-second rates and fine-grained bursts work correctly.
The RateLimitedClient tab wraps reqwest::Client and calls self.limiter.take(1.0).await before every get(). Because the limiter is shared via Arc, cloning the client is cheap and all clones respect the same global rate — important when a connection pool fans out work.
The main tab spawns twenty concurrent tasks against one client configured for five requests per second with a burst of five. The first five fire immediately (draining the initial full bucket), then the remainder are paced out at the refill rate, demonstrating both burst tolerance and steady-state throttling. Pitfalls to watch for: choosing capacity too large permits big bursts that can still trip upstream limits, and clock adjustments can perturb Instant math, though Instant is monotonic and safe here.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
struct Config<'a> {
name: &'a str,
value: &'a str,
}
fn parse_config(line: &str) -> Config {Lifetime annotations for flexible borrowing in structs
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use tracing::{info, instrument};
#[instrument]
fn process_request(user_id: u64) {
info!(user_id, "Processing request");
// Work happens heretracing for structured logging and distributed tracing
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
Share this code
Here's the card — post it anywhere.
