backpressure-pipeline

A personal systems project: a self-contained building block for the common shape “one fast producer → N async workers”, where overload must degrade predictably instead of crashing.

The problem

An unbounded queue applies no back-pressure, so a slow consumer turns a fast producer into an OOM — a crash, not a slowdown. And when a bounded buffer is full, something must be discarded: a position or state stream wants the freshest sample and should drop the oldest; an event or command stream needs order and visibility of loss and should drop the newest and alarm. Leaving that choice implicit gets it wrong. Shutdown has its own trap: detached tasks that are never joined leak work.

The approach

A bounded MPMC hand-off queue built as a Semaphore over a Mutex<VecDeque>, so permit accounting is exact and there is no lost-wakeup reasoning to get wrong. The shed policy (DropNewest / DropOldest) is a required, explicit choice. Shutdown is structured via a CancellationToken tree and a JoinSet; every spawned task is joined, and the run reports workers_joined / aborted so leaks are observable. The conservation law produced == processed + dropped is asserted in the demo and the tests.

What the tests prove

• Fully deterministic proof of each policy’s semantics: drop-newest keeps the oldest items, drop-oldest keeps the newest.
• A worker never exits while items remain in the buffer.
• Under real overload, both policies satisfy produced == processed + dropped and every worker is joined.
• An external cancel stops intake early, yet conservation still holds and all workers are still joined — structured, not detached.

Five tests pass.

Running it

cargo run --bin demo runs the policy illustration, an overload stress, and the conservation check, then exits 0:

== policy illustration: buffer=4, push ids 0..8, then drain ==
  DropNewest keeps [0, 1, 2, 3]  (newest arrivals shed; backlog order preserved)
  DropOldest keeps [4, 5, 6, 7]  (oldest shed; always the freshest state)

== overload stress: 4000-item burst, buffer=16, 4 slow workers ==
policy        produced  processed  dropped   joined  aborted
------------------------------------------------------------
DropNewest        4000         48     3952        4        0
DropOldest        4000        128     3872        4        0
------------------------------------------------------------
conservation held for both policies; no tasks leaked. demo OK.

Source

The full crate — Cargo.toml, src/, tests/, and an inline rendering of every source file — is browsable at /projects/backpressure-pipeline/.