Streams, Backpressure, Buffering, and Fairness

26 MiB is the first useful answer in the relay-service capacity review:

256 connections ×   8 KiB receive allowance =  2 MiB
512 messages    ×  16 KiB decoded envelope   =  8 MiB
 32 workers     × 256 KiB active batch       =  8 MiB
128 commits     ×  64 KiB pending batch      =  8 MiB
                                                   -----
bounded pipeline payload reservation             26 MiB

It is also incomplete. The number excludes allocator fragmentation, container headers, futures, task storage, channel nodes, TLS state, parser scratch space, socket buffers, runtime queues, kernel memory, telemetry, and temporary duplication. Its value is not that it predicts RSS exactly. Its value is that four previously implicit multipliers are now reviewable.

An async pipeline is bounded only when every place that can retain work has a finite capacity and a defined response to saturation. A bounded channel in the middle cannot compensate for unbounded connection admission at the front or an unlimited retry set at the back. Backpressure is therefore an end-to-end ownership property: the consumer’s finite ability to make progress must reach the producer as waiting, reduced demand, refused admission, or deliberate shedding.

The stream contract is pull-shaped, even when events originate elsewhere

Rust’s standard library defines Future, but the commonly used Stream trait lives in the futures-core ecosystem crate. Its essential method is:

fn poll_next(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>>;

The three results form a protocol:

• Ready(Some(item)) transfers one item and permits a later poll for another;
• Pending means no item is ready now and obliges the implementation to arrange a wake when progress may be possible;
• Ready(None) means the stream has terminated; callers must not assume arbitrary further polling is supported.

This is demand, not storage. Polling asks for the next item; it does not say how many items an upstream socket, channel, parser, or broker may already retain. size_hint estimates remaining stream length where meaningful. It is not a buffer reservation and does not impose a memory bound.

Many event sources are push-shaped below this interface. A network driver reports readiness. A broker sends deliveries. Multiple producers enqueue into an MPSC channel. The adapter must reconcile that push with downstream demand. If it keeps reading while the consumer is stalled, the queue moves into the adapter, kernel, peer, or broker. The bytes do not disappear.

Readiness also differs from completion. A sink that is ready for one item may become saturated after accepting it. A socket being writable does not guarantee an entire frame will be written. A consumer that can poll one item does not grant unlimited upstream credit. Protocol design should expose the unit of demand: byte, message, batch, request permit, or durable commit slot.

Draw every retaining edge

For a capacity review, replace an architecture’s arrows with finite reservoirs:

connection permits
  → socket receive allowance
  → decoder queue
  → active worker permits
  → per-worker scratch and batch
  → commit queue
  → storage client inflight set
  → retry/dead-letter reservation

For each edge, name its capacity in items or bytes and measure the p50, p95, and hard maximum retained envelope. Name the component that releases the reservation. State what the producer does at saturation: await, reject, shed, spill, or disconnect. Then trace ownership through cancellation and attach the metric and test that prove occupancy and overload behavior.

Count both queue slots and in-flight work. A channel capacity of 512 plus 32 workers can retain at least 544 message envelopes, and workers may hold expanded decoded forms or output batches larger than the queue item. Fan-out can multiply one input into several retained outputs. Retries can duplicate payloads unless they share or reconstruct data deliberately.

An unbounded queue still has a bound: available memory and eventual process termination. Tokio’s unbounded MPSC documentation states this plainly. Treat “unbounded” as “memory is the overload policy,” which is rarely a defensible default for production data paths.

Backpressure, admission control, and rate limiting solve different problems

Backpressure makes a producer wait or reduce demand because downstream capacity is currently unavailable. Admission control rejects or defers new work before the system takes ownership of it. Rate limiting restricts work over time, even if instantaneous capacity is available. A service often needs all three.

Consider a full decoded-message queue:

• A cooperative internal producer can await capacity. The wait itself must have a cancellation and deadline policy.
• An HTTP endpoint can refuse admission with an overload response before reading a large body.
• A broker consumer can stop requesting deliveries, reduce credit, or negative-ack according to broker semantics.
• A UDP receiver cannot push back through the transport; it must drop, sample, or use an application protocol that tolerates loss.
• A control-plane message may reserve capacity unavailable to bulk traffic.

The saturation action belongs at the boundary that still has a safe choice. Once a service has acknowledged durable ownership, shedding the item is no longer ordinary overload control; it is data loss unless a stronger protocol transfers ownership elsewhere.

Bounded channels provide local backpressure. Tokio’s bounded MPSC waits when its finite buffer is full and reports disconnect when the other side is gone. That is a useful mechanism, not an end-to-end policy. If every socket handler waits on the same full channel while retaining a large request body and connection state, the service has moved the queue into hundreds of suspended tasks.

The companion fixture makes the decision explicit:

pub fn overload_action(
    occupied: usize,
    capacity: usize,
    may_wait: bool,
) -> OverloadAction {
    if occupied < capacity {
        OverloadAction::Admit
    } else if may_wait {
        OverloadAction::Wait
    } else {
        OverloadAction::Shed
    }
}

Real admission also considers reserved capacity, caller class, deadline, payload size, downstream health, and whether work is already committed. The small function’s purpose is to prevent the full-queue case from remaining unspecified.

Buffering trades coordination for retained state

A buffer can smooth short bursts, decouple scheduling jitter, combine small writes, or keep an expensive device busy. It cannot increase sustainable throughput beyond the bottleneck. If arrival rate remains above service rate, every finite buffer eventually fills.

Use queueing delay as part of the request budget. A message that spends 800 ms waiting in a queue before a 300 ms operation cannot satisfy a one-second end-to-end deadline, even if the operation’s local timeout is one second. Chapter 63 will allocate deadlines across attempts; the necessary input is already here: queue age and remaining budget at dequeue.

Capacity should follow a stated burst model and measured envelope, not a round number. A defensible choice might say: “absorb a 200 ms burst at 2,500 messages/s above steady-state service, cap decoded messages at 16 KiB, reserve 8 MiB, and reject before body expansion when the queue or memory watermark is full.” The calculation can be wrong and improved. “Use 10,000 because that seems safe” cannot be audited.

Large buffers can make dashboards look healthy while latency grows invisibly. Small buffers expose the bottleneck earlier but may amplify transient shedding. Evaluate both occupancy and age. A half-full queue of ten-minute-old work is more alarming than a momentarily full queue whose oldest item is two milliseconds old.

Batch size has three separate dimensions

Batching amortizes fixed overhead: syscalls, locks, protocol frames, compression setup, or storage transactions. It also retains more memory and delays the oldest item while the batch fills.

Do not compress batch policy into one number. Define:

1. a maximum item count;
2. a maximum byte count after relevant expansion;
3. a maximum age or remaining-deadline threshold.

The first prevents pathological lists of tiny items. The second handles large envelopes. The third prevents low traffic from waiting indefinitely for a full batch. Processing time and cancellation behavior matter too: if a batch is cancelled after ten of one hundred effects commit, the protocol must identify the completed subset.

Adaptive batches may improve utilization, but their controller needs bounds and stable inputs. Increasing batch size in response to queue depth can worsen tail latency and memory precisely during overload. Measure service time per item, fixed cost per batch, queue age, memory per batch, and downstream concurrency before choosing.

Fan-in and fan-out change the fairness problem

Fan-out distributes work to several consumers. It can increase throughput when the downstream resource scales, but a shared queue does not guarantee that one tenant, key, or expensive item cannot dominate workers. Partitioning by key preserves local ordering but can create hot shards. Work stealing improves utilization but may weaken cache locality and ordering assumptions.

Fan-in merges several producers into one consumer. A naïve merger that drains the first ready source until empty can starve quieter sources whenever the busy source is never empty. The lab’s deterministic model takes at most one ready item from each source per round:

for source in &mut sources {
    if let Some(item) = source.pop_front() {
        merged.push(item);
    }
}

Given a1,a2,a3, b1, and c1,c2, it produces a1,b1,c1,a2,c2,a3. That proves the model’s order, not Tokio’s scheduler fairness and not global fairness across sockets, runtime workers, locks, storage, and the network.

Fairness must name a population and a resource. Possibilities include equal turns per ready source, weighted service by tenant, deficit round robin by bytes, oldest-deadline-first, reserved concurrency for control traffic, or a maximum consecutive batch from one partition. Equal item counts are not fair when one item costs a thousand times another.

Measure progress, not merely distribution at admission:

• service count and bytes by tenant, partition, or priority class;
• wait-time and queue-age distributions by class;
• longest interval without service for a continuously ready source;
• worker occupancy by item cost;
• deadline misses and shedding by class;
• reordering where the protocol promises order.

Runtime selection primitives may include fairness behavior, but it is local and version-specific. A fair semaphore does not make an entire pipeline fair. A scheduler that eventually polls tasks cannot repair head-of-line blocking inside one task.

Head-of-line blocking hides inside valid bounds

A queue can be perfectly bounded and still provide unacceptable progress. One slow item at the front can delay every item behind it when ordering is strict. One connection can monopolize a parser loop. One storage batch containing a hot key can block unrelated keys.

Repairs alter semantics and cost:

• partition queues by independence key, accepting per-partition imbalance;
• cap work per poll or per turn, adding scheduling overhead;
• split large items, requiring resumable state and partial-failure handling;
• run bounded concurrent operations, weakening completion order;
• reserve lanes for latency-sensitive traffic, reducing bulk utilization;
• reject items whose estimated cost exceeds the remaining deadline.

Do not promise fairness by inserting yield_now() after an arbitrary count. Yielding can improve cooperative scheduling, but the unit of work, runtime behavior, and downstream bottleneck still determine progress. A measured quantum tied to bytes or CPU time is stronger evidence.

Synchronous bridges need capacity on both sides

Async code eventually meets synchronous code: compression libraries, filesystem APIs, database drivers, legacy callbacks, or CPU-heavy transforms. The bridge often consists of an async queue, a blocking pool, and a result queue. All three retain work.

If the async side submits faster than the blocking side completes, an unlimited blocking-task queue defeats the bounded async channel. Acquire a finite permit before submission, include queued blocking work in the memory budget, and define what cancellation means after synchronous work starts. Aborting the async waiter may not stop the underlying blocking operation.

The reverse bridge matters too. A synchronous callback that uses an unbounded sender because it cannot await has chosen memory exhaustion as pressure relief. Alternatives include a bounded nonblocking try_send with an explicit loss counter, a dedicated bounded thread queue, source-level flow control, or refusing integration when the source cannot meet the reliability contract.

Bridge metrics should separate time waiting for a permit, time queued for a blocking worker, execution time, result-queue wait, and abandoned-result count. One “operation latency” histogram cannot locate saturation.

Calculate the whole budget, then verify RSS

The lab represents each finite payload reservation as capacity times retained bytes:

pub struct Reservation {
    pub capacity: usize,
    pub bytes_per_item: usize,
}

pub fn payload_budget(stages: &[Reservation]) -> usize {
    stages.iter().map(|stage| stage.bytes()).sum()
}

Its test proves the visual’s four reservations sum to 26 MiB. A production budget expands the ledger:

payload reservations
+ message and collection overhead
+ parser and transform scratch
+ task/future and channel bookkeeping
+ connection, TLS, and protocol state
+ runtime and blocking queues
+ userspace socket buffers
+ expected allocator fragmentation
+ telemetry buffers
+ transient duplication during batch/encode/retry
= process userspace target

Kernel socket buffers and page cache may be operationally charged to the service even when they do not appear in the same RSS measure. Container limits, cgroup accounting, and allocator behavior are platform-specific. State the measurement boundary.

Use both calculation and experiment. Generate envelopes at the declared maximum, fill every queue, hold workers at their peak scratch allocation, trigger retry retention, and observe RSS plus allocator and kernel metrics. Then release work and verify memory returns to an expected steady-state range. A bound that exists only in configuration but not under the load generator is not evidence.

Instrument the pressure path

Every finite reservoir needs occupancy, capacity, and age. Every admission point needs accepted, waited, rejected, shed, and disconnected counts with reasons. Every worker pool needs active permits, queue wait, service time, and completion outcome. Record bytes as well as items.

Useful alert relationships include:

• queue age consumes a material fraction of the oldest item’s remaining deadline;
• occupancy remains above a watermark longer than the designed burst window;
• producers spend increasing time waiting for capacity while consumer throughput is flat;
• shedding rises but downstream utilization is low, suggesting a misplaced bound;
• one tenant’s longest no-progress interval grows while aggregate throughput looks healthy;
• process memory exceeds the calculated reservation by a widening or unexplained margin;
• a synchronous bridge’s pending work grows after async queues stabilize.

High occupancy is not automatically failure. A deliberately batched commit queue may operate efficiently near capacity. The alert should reflect age, throughput, deadline, memory, and rejection policy together.

Exercise: defend relay-service under overload

Set an end-to-end memory and progress budget for a deployment with a 256 MiB userspace target. Begin with the 26 MiB payload reservation, but do not simply subtract it and declare victory.

Produce these artifacts:

1. A retention map from connection admission through durable storage and retry. Include queue slots, in-flight work, futures, parser scratch, batch expansion, socket/TLS state, and transient copies.
2. An envelope table with typical, p95, and hard maximum bytes. Multiply hard or justified statistical bounds by each capacity, and state the safety margin for allocator/runtime overhead.
3. A saturation policy for every edge: wait, reduce demand, reject, shed, spill, or disconnect. Mark the last boundary at which the service can refuse work without violating an acknowledgment.
4. A batching policy with count, bytes, and age limits. Show how a partial commit and cancellation are reconciled.
5. A fairness experiment with one continuously busy tenant, ten low-rate tenants, and a mixture of cheap and expensive messages. Define the maximum acceptable no-progress interval and compare FIFO, per-tenant round robin, and weighted byte-based service.
6. A synchronous bridge design with finite submission permits and explicit behavior when the blocking side cannot be cancelled.
7. A load test that fills all declared reservoirs, verifies wait/rejection counters, compares calculated memory with observed accounting, and confirms recovery after load stops.

A strong answer may change the visual’s capacities. The exercise is not to preserve 26 MiB; it is to make every multiplier and overload choice defensible. Reject designs that call an unbounded retry queue “temporary,” count only channel payloads, or infer fairness from aggregate throughput.

Pipeline review in one pass

• Every retaining edge has finite item and byte bounds.
• Queue capacity includes in-flight workers, batches, retries, and bridge queues.
• The stream polling contract is not confused with storage capacity.
• Backpressure, admission control, and rate limiting have distinct roles.
• Full-queue behavior is explicit at the boundary that still owns the choice.
• Batch count, bytes, and age are bounded separately.
• Fan-in/fan-out policy names ordering and fairness scope.
• Head-of-line blocking is measured by class and longest no-progress interval.
• Sync/async bridges acquire finite capacity before submitting work.
• Memory calculations state what they omit and are checked against observed accounting.
• Queue telemetry includes age and bytes, not only item count.
• Cancellation preserves ownership of dequeued or partially committed work.

With finite reservoirs, overload becomes a controlled response rather than a memory surprise. The remaining question is temporal: how much of a caller’s deadline may be spent waiting here, how many attempts fit after that wait, and who cleans up the losing operations? Those are the time and idempotency contracts of Chapter 63.

Sources and version note

The stream protocol follows the official crate documentation for futures_core::Stream, including Pending, termination, wake registration, and the limits of size_hint. Tokio’s mpsc::channel documents finite buffering, waiting on full capacity, ordering, and disconnect behavior; mpsc::unbounded_channel explicitly places the effective bound at available system memory. These crate policies are current dependency behavior, not Rust language guarantees and not a promise of global scheduler fairness.

The task-pipeline-lab lockfile records the resolved Tokio and Tokio Utilities versions, declares Rust 1.85 as MSRV, and uses deterministic data structures only for its fairness model. Its 26 MiB test is exact for the four payload reservations and intentionally excludes whole-process overhead. Revalidate the budget assumptions and version-sensitive claims whenever the dependencies or deployment shape change.