Backpressure Across Queues

Key Points

• Backpressure is the signal a slow consumer sends upstream to stop overwhelming it. Without it, queues grow unbounded → OOM, GC stalls, or latency cliffs.
• Mechanism families: bounded buffers (Channels, Dataflow), prefetch / basic.qos / PrefetchCount / max.poll.records, credit-based flow control (Reactive Streams), TCP-style windowing.
• Pull vs push: Kafka is pull-based — natural backpressure (consumer asks for the next batch). Service Bus / RabbitMQ are push-based — need explicit prefetch / concurrency caps.
• Reactive Streams spec standardized async backpressure (Java's Flow API, Project Reactor, Akka Streams, RxJava). .NET equivalents: System.Threading.Channels, TPL Dataflow, IAsyncEnumerable with bounded buffering.
• Stage-by-stage flow control: a slow stage applies pressure to all upstream stages by virtue of bounded buffers. The whole pipeline runs at the speed of the slowest stage.
• Observability: queue depth, consumer lag, ingest rate vs processing rate. If the gap widens, the system is unhealthy long before it crashes.

Concepts (deep dive)

What "no backpressure" looks like

   producer ──► [unbounded queue] ──► consumer
   1000 msg/s                          200 msg/s

Difference per second: 800 messages. Per minute: 48,000. Per hour: 2.88M. Memory grows linearly until:

• Process OOMs.
• GC pauses lengthen, latency spikes.
• The host swaps, and everything tanks together.

There is no version of this that "catches up" — the queue is a leak, not a buffer.

What backpressure does

   producer ──► [bounded queue capacity=N] ──► consumer
                                              │
                                              ▼ slow
   producer's writes BLOCK / FAIL / DROP when queue full

The choice — block, fail, or drop — is the design decision. Each has a name.

Bounded queues in .NET: System.Threading.Channels

var ch = Channel.CreateBounded<Job>(new BoundedChannelOptions(capacity: 1000)
{
    FullMode = BoundedChannelFullMode.Wait,        // producer waits when full
    SingleReader = false,
    SingleWriter = false
});

FullMode options:

// Producer
await ch.Writer.WriteAsync(job, ct);   // awaits when bounded queue full

// Consumer
await foreach (var job in ch.Reader.ReadAllAsync(ct))
    await Process(job, ct);

If the producer is doing the awaiting, the upstream of the producer feels the slowdown — that's how pressure propagates.

TPL Dataflow with bounded blocks

var transform = new TransformBlock<Order, Invoice>(
    o => CreateInvoice(o),
    new ExecutionDataflowBlockOptions
    {
        BoundedCapacity = 500,
        MaxDegreeOfParallelism = 8
    });

var save = new ActionBlock<Invoice>(
    i => Save(i),
    new ExecutionDataflowBlockOptions { BoundedCapacity = 100 });

transform.LinkTo(save, new DataflowLinkOptions { PropagateCompletion = true });

BoundedCapacity is the per-block buffer size. When save is full, transform can't post to it; transform fills its own buffer; eventually transform.SendAsync(...) from the source blocks. Pressure flows backward stage by stage.

Reactive Streams (the spec)

A Java standard, ported / inspired across stacks (Project Reactor, RxJava, Akka Streams). The core abstraction: the consumer requests N items from the producer.

Subscriber.onSubscribe(Subscription s) {
    s.request(64);   // I can take 64 items right now
}
Subscriber.onNext(item) { ... ; if (almost done) s.request(64); }

This is demand-driven — producer only emits what's been requested. .NET hasn't standardized this exactly, but Channels and IAsyncEnumerable capture similar semantics: the consumer's await foreach pull rate gates the producer.

Project Reactor / RxJava operators (onBackpressureBuffer, onBackpressureDrop, onBackpressureLatest) are the same FullMode choices in different clothes.

Pull vs push messaging

Pull (Kafka)

consumer: poll(100 messages)
broker:   here are 100, your offset is now X
consumer: ...processes...
consumer: poll(100)
broker:   here are 100, offset Y

The consumer controls cadence. If processing slows, polls slow. The broker doesn't push anything that wasn't asked for. Backpressure is structural.

Knobs: - max.poll.records: batch size cap. - max.poll.interval.ms: how long between polls before the consumer is considered dead. - fetch.max.bytes: bytes per fetch request.

If processing exceeds max.poll.interval.ms, the broker rebalances the consumer out — your real throughput limit is a function of poll interval and per-batch processing time.

Push (Service Bus, RabbitMQ classic, AMQP basic.deliver)

broker: here's a message  ──► consumer
broker: here's another    ──► consumer (still working on previous)
broker: here's another    ──► consumer
[unbounded local buffering or prefetch cap saves you]

Without a prefetch cap, the broker can shove more messages than the consumer can hold.

Service Bus prefetch

var processor = client.CreateProcessor(queue, new ServiceBusProcessorOptions
{
    PrefetchCount = 20,                 // how many to pull eagerly
    MaxConcurrentCalls = 8,             // parallelism
    AutoCompleteMessages = false        // explicit settle
});

Prefetch is a local cache; messages above MaxConcurrentCalls wait in memory until a worker is free. Set PrefetchCount modestly (2–3× MaxConcurrentCalls) — bigger doesn't help once you saturate workers, and it locks messages (reducing redelivery agility).

RabbitMQ basic.qos

channel.BasicQos(prefetchSize: 0, prefetchCount: 32, global: false);

prefetchCount is the credit limit — broker won't deliver more than this many unacked messages per consumer. Critical: without it, RabbitMQ pushes everything it has, and one slow consumer hoards the queue.

Kafka max.poll.records

max.poll.records=500
max.poll.interval.ms=300000

Pull-based, so technically you control this anyway. Tune max.poll.records to a batch your consumer can finish in << max.poll.interval.ms.

Credit-based flow control

The general primitive behind both Reactive Streams and AMQP basic.qos:

consumer initially grants the broker N credits
broker delivers messages, decrementing credits
when credit hits 0, broker stops
consumer acks → grants more credits → broker resumes

Compare to TCP: receive-window advertises bytes-of-buffer-available; sender can't outpace it.

Push to pull conversion

Sometimes you have a push source and want pull semantics. The pattern: push into a bounded Channel, consume with await foreach.

processor.ProcessMessageAsync += async args =>
{
    if (!_inbox.Writer.TryWrite(args.Message))
    {
        // Queue full → don't complete, let it redeliver later
        await args.AbandonMessageAsync(args.Message);
        return;
    }
    await args.CompleteMessageAsync(args.Message);
};

// Worker
await foreach (var msg in _inbox.Reader.ReadAllAsync(ct))
    await Handle(msg, ct);

Caveat: completing the message before processing it = at-most-once. Real systems either (a) hold the message, process from the channel synchronously with a result, then complete, or (b) accept at-least-once and have idempotent handlers. See Idempotency Keys — Deep.

HttpClient + Polly: client-side backpressure

When you're the producer of HTTP traffic, downstream rate limits and 429s are backpressure signals.

var pipeline = new ResiliencePipelineBuilder<HttpResponseMessage>()
    .AddRateLimiter(new SlidingWindowRateLimiter(new()
    {
        PermitLimit = 100,
        Window = TimeSpan.FromSeconds(1),
        QueueLimit = 200,
        QueueProcessingOrder = QueueProcessingOrder.OldestFirst
    }))
    .AddRetry(new()
    {
        ShouldHandle = new PredicateBuilder<HttpResponseMessage>()
            .HandleResult(r => r.StatusCode == HttpStatusCode.TooManyRequests),
        DelayGenerator = args =>
            ValueTask.FromResult<TimeSpan?>(
                args.Outcome.Result?.Headers.RetryAfter?.Delta ?? TimeSpan.FromSeconds(1))
    })
    .Build();

429 + Retry-After is the server saying "back off". Respect it.

Stage-by-stage pipeline example

[ingest]  →  Channel(cap=1000)  →  [parse]  →  Channel(cap=500)  →  [enrich]  →  Channel(cap=200)  →  [save]
 4000/s                              3500/s                          1500/s                            500/s

save is the bottleneck at 500/s. Pressure cascades: 1. enrich's output channel (cap=200) fills. 2. enrich blocks on WriteAsync. 3. parse's output channel (cap=500) fills. 4. parse blocks. Etc.

Eventually ingest's producer can't enqueue → it must drop, fail, or block its caller. The system runs at 500/s end-to-end; nothing further accumulates.

This is the desired property. The whole pipeline is self-throttling.

Observable signals

You should be alerting on the gap between ingest rate and processing rate.

• Queue depth: messages currently waiting. Trend matters more than instantaneous value.
• Consumer lag (Kafka): (latest_offset - committed_offset) per partition. Growing lag = consumer can't keep up.
• Service Bus active message count: ARM metric.
• Channel-based pipelines: instrument with Meter / Counter for items in / out per stage; compute deltas.

var counter = meter.CreateCounter<long>("pipeline.items.processed");
counter.Add(1, new("stage", "save"));

Pair with histogram for per-stage latency. Diverging in vs out = the early warning that something will break.

Anti-patterns

// ❌ unbounded queue
var jobs = new ConcurrentQueue<Job>();   // no cap → OOM

// ❌ fire-and-forget
foreach (var item in source) _ = Task.Run(() => Process(item));   // no concurrency limit

// ❌ swallowed task failures
TaskScheduler.UnobservedTaskException += (s,e) => e.SetObserved();   // hiding bugs

Task.Run in a loop without SemaphoreSlim / Parallel.ForEachAsync is the silent OOM. The thread pool will queue them; memory grows; eventually the process dies.

Replace with bounded async parallelism

await Parallel.ForEachAsync(source, new ParallelOptions
{
    MaxDegreeOfParallelism = 16,
    CancellationToken = ct
}, async (item, ct) => await Process(item, ct));

Or:

var sem = new SemaphoreSlim(16);
var tasks = source.Select(async item =>
{
    await sem.WaitAsync(ct);
    try { await Process(item, ct); }
    finally { sem.Release(); }
});
await Task.WhenAll(tasks);

Producer-consumer with bounded channel — full example

public sealed class IngestPipeline
{
    private readonly Channel<RawEvent>     _stage1 = Channel.CreateBounded<RawEvent>(new BoundedChannelOptions(1000) { FullMode = BoundedChannelFullMode.Wait });
    private readonly Channel<ParsedEvent>  _stage2 = Channel.CreateBounded<ParsedEvent>(new BoundedChannelOptions(500)  { FullMode = BoundedChannelFullMode.Wait });
    private readonly Channel<EnrichedEvent>_stage3 = Channel.CreateBounded<EnrichedEvent>(new BoundedChannelOptions(200) { FullMode = BoundedChannelFullMode.Wait });

    public ValueTask EnqueueAsync(RawEvent e, CancellationToken ct) =>
        _stage1.Writer.WriteAsync(e, ct);   // backpressure when full

    public Task RunAsync(CancellationToken ct) => Task.WhenAll(
        Worker(_stage1.Reader, _stage2.Writer, Parse,    ct),
        Worker(_stage2.Reader, _stage3.Writer, Enrich,   ct),
        Worker(_stage3.Reader, sink: SaveAsync,          ct));

    private static async Task Worker<TIn, TOut>(
        ChannelReader<TIn> reader, ChannelWriter<TOut> writer,
        Func<TIn, ValueTask<TOut>> step, CancellationToken ct)
    {
        await foreach (var item in reader.ReadAllAsync(ct))
        {
            var next = await step(item);
            await writer.WriteAsync(next, ct);   // applies pressure upstream
        }
        writer.Complete();
    }

    private static async Task Worker<T>(
        ChannelReader<T> reader, Func<T, ValueTask> sink, CancellationToken ct)
    {
        await foreach (var item in reader.ReadAllAsync(ct))
            await sink(item);
    }

    private static ValueTask<ParsedEvent>   Parse(RawEvent e)         => /* ... */ default;
    private static ValueTask<EnrichedEvent> Enrich(ParsedEvent e)     => /* ... */ default;
    private static ValueTask                SaveAsync(EnrichedEvent e) => /* ... */ default;
}

A slow SaveAsync propagates: stage3 channel fills → enrich blocks → stage2 fills → parse blocks → stage1 fills → producer's EnqueueAsync awaits. Nothing accumulates beyond the configured capacities.

How it works under the hood

What "blocking" means in async

WriteAsync on a full bounded channel doesn't block a thread — it returns a pending ValueTask. The producer's continuation only runs when a slot opens. Threads stay free; the logical operation waits.

This is why bounded async backpressure is cheap: no thread is held hostage.

AMQP credit cycle

client → broker:   basic.consume (ack-mode=manual)
client → broker:   basic.qos prefetch=32
broker → client:   basic.deliver msg-1
broker → client:   basic.deliver msg-2
... up to 32 unacked
client → broker:   basic.ack msg-1     (credit returned → broker can deliver one more)
broker → client:   basic.deliver msg-33

Without basic.qos, prefetchCount=0 = unlimited. RabbitMQ will deliver as fast as the TCP socket allows.

Kafka offsets and lag

partition log:  [0][1][2][3][4][5][6][7][8][9]
                                         ▲      ▲
                                         │      latest offset = 9
                                         consumer committed = 6
                                         lag = 3

Lag growing over time = consumer falling behind. Per-partition metric; alert on max-across-partitions.

Code: correct vs wrong

❌ Wrong: unbounded queue

var q = new ConcurrentQueue<Event>();
producer.OnEvent(e => q.Enqueue(e));
// no consumer → OOM ; consumer too slow → OOM

✅ Correct: bounded channel

var ch = Channel.CreateBounded<Event>(new BoundedChannelOptions(10_000)
{
    FullMode = BoundedChannelFullMode.Wait
});

❌ Wrong: fire-and-forget per item

foreach (var item in source) _ = Task.Run(() => Heavy(item));

✅ Correct: bounded parallelism

await Parallel.ForEachAsync(source, new() { MaxDegreeOfParallelism = 16 },
    async (i, ct) => await Heavy(i, ct));

❌ Wrong: RabbitMQ consumer with no basic.qos

channel.BasicConsume(queue, autoAck: false, consumer);
// broker pushes everything; one slow consumer hoards the queue

✅ Correct: prefetch

channel.BasicQos(prefetchSize: 0, prefetchCount: 32, global: false);
channel.BasicConsume(queue, autoAck: false, consumer);

❌ Wrong: Service Bus MaxConcurrentCalls = 1000 with 1 worker

new ServiceBusProcessorOptions { MaxConcurrentCalls = 1000 }   // not what you think

The processor will hold up to 1000 messages in flight → one container, GC choke.

✅ Correct: tune to actual capacity

new ServiceBusProcessorOptions
{
    MaxConcurrentCalls = 16,    // matches your worker capacity
    PrefetchCount = 32          // 2× MaxConcurrentCalls
}

❌ Wrong: ignore 429 from a downstream API

while (true) await http.PostAsync(url, body);   // 429 storm; no respect for Retry-After

✅ Correct: rate limit + retry-after

.AddRateLimiter(new SlidingWindowRateLimiter(new() { PermitLimit = 100, Window = TimeSpan.FromSeconds(1) }))
.AddRetry(new() { DelayGenerator = a => ValueTask.FromResult<TimeSpan?>(a.Outcome.Result?.Headers.RetryAfter?.Delta) })

Design patterns for this topic

Pattern 1 — "Bounded Channel between stages"

• Intent: propagate pressure via blocking WriteAsync.

Pattern 2 — "Prefetch + concurrency cap on push brokers"

• Intent: turn a push source into a credit-controlled flow.

Pattern 3 — "Drop-newest / drop-oldest for lossy ingest"

• Intent: prefer freshness or history when full; never drop silently without a metric.

Pattern 4 — "Pull-based pipeline with IAsyncEnumerable / await foreach"

• Intent: consumer pace gates production.

Pattern 5 — "Lag-based autoscale"

• Intent: scale workers when consumer lag / queue depth exceeds threshold.

Pros & cons / trade-offs

When to use / when to avoid

• Use bounded channels for any in-process pipeline.
• Use prefetch + concurrency caps on Service Bus / RabbitMQ.
• Use pull (Kafka) when you want backpressure for free.
• Use drop-modes only with a counter on dropped items and a clear UX/SLA story.
• Avoid ConcurrentQueue / ConcurrentBag as inter-stage queues (no backpressure).
• Avoid Task.Run per item without bounded parallelism.
• Avoid ignoring Retry-After and 429s — that's the server's backpressure signal.

Interview Q&A

Q1. What is backpressure? A signal from a slow consumer to upstream producers to stop overwhelming it — implemented as bounded buffers, credit-based flow, or pull-based protocols.

Q2. Pull vs push backpressure? Pull (Kafka): consumer asks → backpressure structural. Push (Service Bus, RabbitMQ): broker delivers → need explicit prefetch / concurrency caps.

Q3. What's BoundedChannelFullMode.Wait vs DropOldest? Wait: producer awaits when full (lossless backpressure). DropOldest: oldest item dropped when full (lossy; freshness-favoring).

Q4. Why is ConcurrentQueue dangerous as an inter-stage queue? It's unbounded. Producer-consumer rate mismatch grows it without limit until OOM.

Q5. Service Bus PrefetchCount — what do you set it to? Modest multiple of MaxConcurrentCalls (2–3×). Higher doesn't increase throughput once workers saturate, but locks messages and slows redelivery.

Q6. RabbitMQ basic.qos / prefetchCount? Credit limit per consumer. Without it, broker pushes everything and one slow consumer hoards the queue.

Q7. Kafka consumer lag — what is it? latest_offset - committed_offset. Growing lag = consumer falling behind. Alert when it diverges from steady-state.

Q8. Reactive Streams in .NET — what's the equivalent? System.Threading.Channels, TPL Dataflow with BoundedCapacity, IAsyncEnumerable with await foreach. Same pull / credit semantics, different API.

Q9. How do you handle 429 from a downstream? Respect Retry-After; combine client-side rate limiting (Polly AddRateLimiter) with retry-on-429 to avoid slamming during recovery.

Q10. Drop-newest vs drop-oldest — when? Drop-newest if recent state is more valuable (latest sensor reading replaces queued one). Drop-oldest if you're computing something order-sensitive on recent history.

Q11. Stage-by-stage flow control — why does it work? Each bounded buffer creates a chokepoint. Slow stage fills its input buffer, blocking the previous stage's output — pressure cascades upstream by purely local rules.

Q12. Most common .NET backpressure mistake? Task.Run per item with no concurrency limit, and ConcurrentQueue as a hand-rolled inter-stage buffer. Both produce unbounded growth.

Gotchas / common mistakes

• ⚠️ Unbounded queues anywhere — silent OOM in production.
• ⚠️ Task.Run in loops without SemaphoreSlim / Parallel.ForEachAsync.
• ⚠️ Service Bus MaxConcurrentCalls set to a huge number — holds messages, GC chokes.
• ⚠️ RabbitMQ without basic.qos — broker over-delivers; one slow consumer ruins the queue.
• ⚠️ Kafka max.poll.records larger than what fits in max.poll.interval.ms — rebalances; loops.
• ⚠️ Drop modes without a counter — silent data loss.
• ⚠️ Treating 429s as transient errors — retry storms make it worse.
• ⚠️ Swallowing UnobservedTaskException — hides backpressure failure modes.
• ⚠️ Not measuring queue depth and lag — first sign of trouble is a crash.

Further reading

• MS Learn: System.Threading.Channels
• MS Learn: TPL Dataflow
• MS Learn: Service Bus prefetch
• Reactive Streams spec
• Kafka consumer config
• Stephen Toub: Channels in .NET