Backpressure in Rust Pipelines and Services Explained

A system is a set of stages connected by queues: accept requests, parse, process, write. If any stage is slower than the one feeding it, the queue in front of it grows. With unbounded queues it grows until the process is killed. Backpressure is the mechanism that connects the queue's fullness back to the producer. In Rust it is mostly a matter of choosing bounded types. Versions: Rust 1.90, Tokio 1.47, futures 0.3.

The failure without it

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::unbounded_channel::<Vec<u8>>();

    tokio::spawn(async move {
        loop {
            // producer: no pacing, produces as fast as it can
            if tx.send(vec![0u8; 1024]).is_err() {
                break;
            }
            tokio::task::yield_now().await;
        }
    });

    while let Some(item) = rx.recv().await {
        // consumer: slower than the producer
        tokio::time::sleep(std::time::Duration::from_millis(1)).await;
        drop(item);
    }
}

The producer runs at thousands of items per second and the consumer at about a thousand. The difference accumulates in the channel at megabytes per second, with no error until the allocator fails. The program has no way to express "I am full".

Bounded channel: the basic form

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<Vec<u8>>(100);

    let producer = tokio::spawn(async move {
        for i in 0..10_000 {
            // waits here when 100 items are queued: this IS the backpressure
            if tx.send(vec![i as u8; 1024]).await.is_err() {
                break;
            }
        }
    });

    while let Some(item) = rx.recv().await {
        tokio::time::sleep(std::time::Duration::from_millis(1)).await;
        drop(item);
    }
    producer.await.unwrap();
}

When the queue holds 100 items, send().await suspends the producer until the consumer takes one. Memory is bounded by 100 items, and the producer's speed matches the consumer's. The same applies across a pipeline: each stage uses a bounded channel to the next, and pressure propagates upstream to the source (bounded channels in Tokio).

Choosing the full-queue policy

When the queue is full, something has to give:

| Policy | API | Use when | |---|---|---| | Wait | send().await | the producer can pause (internal pipelines, file readers) | | Reject newest | try_send returning Full | you can tell the caller to retry (HTTP: 503) | | Drop oldest | ring buffer, broadcast lag | only fresh data matters (metrics, telemetry) | | Drop newest | try_send and ignore | best-effort logging | | Error | try_send and propagate | upstream must know |

use tokio::sync::mpsc::error::TrySendError;

fn enqueue(tx: &mpsc::Sender<Job>, job: Job) -> Result<(), &'static str> {
    match tx.try_send(job) {
        Ok(()) => Ok(()),
        Err(TrySendError::Full(_)) => Err("busy"),       // caller returns 503 with Retry-After
        Err(TrySendError::Closed(_)) => Err("shutting down"),
    }
}

In a web handler, waiting on a full queue holds the request open and ties up resources, so reject quickly (try_send) and let the client back off (graceful degradation).

Limit in-flight work with a semaphore

Backpressure also applies to spawning tasks. A loop that tokio::spawns per incoming item has no bound:

use std::sync::Arc;
use tokio::sync::Semaphore;

async fn process_all(items: Vec<u64>) {
    let permits = Arc::new(Semaphore::new(32));
    let mut handles = Vec::new();
    for item in items {
        // waits here when 32 tasks are running
        let permit = Arc::clone(&permits).acquire_owned().await.unwrap();
        handles.push(tokio::spawn(async move {
            do_work(item).await;
            drop(permit);
        }));
    }
    for h in handles {
        let _ = h.await;
    }
}

async fn do_work(_n: u64) {}

Acquiring the permit before spawning is the key: the loop itself is paced. See semaphores.

Streams: buffer_unordered

use futures::{stream, StreamExt};

async fn fetch(url: String) -> usize {
    url.len()
}

#[tokio::main]
async fn main() {
    let urls: Vec<String> = (0..1000).map(|i| format!("https://example.test/{i}")).collect();
    let total: usize = stream::iter(urls)
        .map(fetch)
        .buffer_unordered(16)          // at most 16 futures in flight
        .fold(0, |acc, n| async move { acc + n })
        .await;
    println!("{total}");
}

buffer_unordered(16) polls the source only when fewer than 16 futures are running, so a lazy source is read at the pace of completion (buffer_unordered, async streams). The stream is pull-based, which gives backpressure by construction: nothing is produced until the consumer asks.

At the network edge

TCP has flow control. If your handler stops reading a socket, the kernel buffer fills, the advertised window shrinks and the sender slows down. This only works if your code actually stops reading: an accept loop that reads requests into an unbounded queue defeats it. Read, process, then read again, with a bound in between. For the accept loop itself, acquire a permit before accept() so the kernel's listen backlog absorbs overflow (async TCP with Tokio). HTTP/2 has per-stream and per-connection windows with the same effect.

Detecting missing backpressure

• Memory that grows steadily under load and recovers after traffic stops.
• Latency that rises over time at a constant request rate (queue growth).
• A metric of queue length that only goes up. Export tx.capacity() and tx.max_capacity() from channels (Tokio runtime metrics).
• unbounded_channel and spawn inside loops in code review.

Design checklist

1. Every queue is bounded, and the bound is a number someone chose.
2. Every stage has a stated policy for full.
3. The source can be slowed or rejected (client gets 503 or 429, file reader waits).
4. In-flight work has a cap.
5. Timeouts exist, so a stuck consumer does not freeze the pipeline forever.
6. Tests simulate a slow consumer and assert memory or queue length stays bounded.

Related patterns: fan-out and fan-in pipelines, an in-process event bus and the actor model.