use std::time::{Duration, Instant};
pub struct LeakyBucket {
level: f64,
capacity: f64,
rate_per_sec: f64,
last_leak: Instant,
}
impl LeakyBucket {
pub fn new(capacity: f64, rate_per_sec: f64) -> Self {
LeakyBucket {
level: 0.0,
capacity,
rate_per_sec,
last_leak: Instant::now(),
}
}
fn leak(&mut self, now: Instant) {
let elapsed = now.duration_since(self.last_leak).as_secs_f64();
let drained = elapsed * self.rate_per_sec;
self.level = (self.level - drained).max(0.0);
self.last_leak = now;
}
pub fn try_add(&mut self, now: Instant) -> bool {
self.leak(now);
if self.level + 1.0 <= self.capacity {
self.level += 1.0;
true
} else {
false
}
}
pub fn time_until_available(&mut self, now: Instant) -> Duration {
self.leak(now);
let overflow = (self.level + 1.0) - self.capacity;
if overflow <= 0.0 {
return Duration::ZERO;
}
Duration::from_secs_f64(overflow / self.rate_per_sec)
}
}
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::Mutex;
use tokio::time::sleep;
use crate::leaky_bucket::LeakyBucket;
#[derive(Clone)]
pub struct RateShaper {
bucket: Arc<Mutex<LeakyBucket>>,
}
impl RateShaper {
pub fn new(capacity: f64, rate_per_sec: f64) -> Self {
RateShaper {
bucket: Arc::new(Mutex::new(LeakyBucket::new(capacity, rate_per_sec))),
}
}
pub async fn acquire(&self) {
loop {
let wait = {
let mut bucket = self.bucket.lock().await;
let now = Instant::now();
if bucket.try_add(now) {
return;
}
bucket.time_until_available(now)
}; // guard dropped here, before awaiting
sleep(wait).await;
}
}
}
use tokio::sync::mpsc::Receiver;
use crate::shaper::RateShaper;
pub struct Message {
pub id: u64,
pub payload: Vec<u8>,
}
pub async fn run_consumer(mut rx: Receiver<Message>, shaper: RateShaper) {
while let Some(msg) = rx.recv().await {
shaper.acquire().await;
if let Err(err) = handle(&msg).await {
eprintln!("failed to process message {}: {}", msg.id, err);
}
}
}
async fn handle(msg: &Message) -> Result<(), Box<dyn std::error::Error>> {
// downstream call paced by the shaper
println!("processing message {} ({} bytes)", msg.id, msg.payload.len());
Ok(())
}
#[tokio::main]
async fn main() {
let (tx, rx) = tokio::sync::mpsc::channel(1024);
let shaper = RateShaper::new(20.0, 10.0); // burst 20, drain 10/s
let consumer = tokio::spawn(run_consumer(rx, shaper));
for id in 0..100 {
let _ = tx.send(Message { id, payload: vec![0u8; 64] }).await;
}
drop(tx);
let _ = consumer.await;
}
This snippet shapes the throughput of an async message consumer using a classic leaky-bucket algorithm. Unlike a token bucket that permits bursts up to its capacity, a leaky bucket enforces a smooth, constant drain rate: the bucket holds queued "water" (pending work), and it leaks at a fixed rate no matter how fast messages arrive. That property makes it a good fit for protecting a fragile downstream — a database, a third-party API, or a socket — from spikes while still absorbing short backlogs up to a bounded capacity.
In leaky_bucket.rs, the LeakyBucket tracks level (current water) and capacity, and computes leakage lazily. Rather than spawning a timer, leak measures elapsed wall-clock time since last_leak and subtracts rate_per_sec * elapsed, clamping at zero. try_add first leaks, then admits one unit only if there is room, returning false when the bucket is full — this is the immediate drop path. time_until_available computes how long a caller must wait for one unit of headroom, which is what turns a hard drop into graceful backpressure. Keeping the math time-based means accuracy doesn't depend on how often the bucket is polled.
shaper.rs wraps the bucket in a Mutex inside a RateShaper so it can be shared across tasks via Arc. Its acquire method loops: it locks, tries to add, and if the bucket is full it reads time_until_available, releases the lock (dropping the guard before awaiting is essential to avoid holding the mutex across .await), and sleeps with tokio::time::sleep. This yields cooperatively so other tasks make progress. The loop then retries, guaranteeing eventual admission at the shaped rate.
consumer.rs ties it together: run_consumer pulls from an mpsc::Receiver, calls shaper.acquire().await before every handle call, and thereby paces processing to the configured rate. Because the shaper is Arc-shared, multiple consumer tasks would share one global rate limit rather than each getting their own. The main trade-off is latency: shaping deliberately delays work to protect the downstream, so capacity should be tuned to tolerate expected bursts without unbounded queue growth. A developer reaches for this pattern whenever a consumer can outrun what it feeds.
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
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
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 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.
