Building Distributed Systems in Elixir: Part 9 - Backpressure

In the previous part of this series, we built a Publish / Subscribe Broker from scratch. We saw how a central coordinator could fan out events across multiple topics to concurrent subscriber processes, completely decoupling publishers from subscribers: Publisher ---> {:publish, :orders, event} ---> [ Broker ] | +-----------------------------+-----------------------------+ | | | {:broadcast, ...} {:broadcast, ...} {:broadcast, ...} v v v [ Inventory Worker ] [ Email Notifier ] [ Analytics Logger ] Broadcasting with send/2 in Elixir feels like magic: you can dispatch tens of thousands of messages in a fraction of a millisecond. However, that non-blocking speed conceals one of the most dangerous failure modes in distributed systems: What happens if the publisher emits 10,000 messages per second, but the email notifier can only send 50 emails per second? In this part, we will explore Backpressure from first principles using raw process primitives: spawn/1 send/2 receive/1 Process.info/2 No GenStage . No Broadway . No Flow . We will watch a BEAM process mailbox explode in real time, examine why unbounded message queues lead to catastrophic Out-Of-Memory (OOM) crashes, and implement two fundamental flow control solutions: Stop-and-Wait and credit-based Demand-Driven Windowing (the exact mechanism powering GenStage ). The Hidden Trap: Unbounded Process Mailboxes In Elixir and Erlang, every process has its own private heap and its own private mailbox. By default, BEAM process mailboxes are unbounded FIFO queues. When Process A sends a message to Process B: send(target_pid, {:work, data}) Two critical things happen: - The message payload is copied to Process B's heap (unless it is a refc binary $> 64$ bytes). - The message is placed at the tail of Process B's mailbox queue. - send/2 immediately returns{:work, data} without waiting for Process B to acknowledge, inspect, or process the message. Because send/2 never blocks, an eager producer receives zero feedback about whether the consumer is keeping up! Fast Producer (10,000 msgs/sec) =====> [ Mailbox: 10k... 50k... 100k ] =====> Slow Consumer (100 msgs/sec) ^ | Unbounded Memory Growth GC overhead degrades node Risk of VM OOM Crash If the producer outpaces the consumer over a sustained period: - Mailbox Explosion: Unhandled messages pile up in the consumer's message queue. - Heap Expansion: The BEAM runtime continuously allocates memory to store the growing queue. - Garbage Collection Thrashing: The BEAM's generational GC must repeatedly inspect the growing queue during garbage collection passes, consuming massive amounts of CPU. - OOM Killer: Eventually, the host machine runs out of physical RAM. The Linux kernel's Out-Of-Memory (OOM) killer wakes up and terminates the entire BEAM OS process ( beam.smp ). To build resilient distributed systems, we need flow control-a mechanism that allows downstream consumers to apply backpressure on upstream producers. Experiment 1: Watching a Mailbox Explode in Real Time Let's build a minimal script that demonstrates this exact failure mode. We'll create: - A Slow Consumer that simulates real-world processing latency (e.g. database writes or external HTTP calls) with Process.sleep(15) . - A Fast Producer that blasts 100 items into the consumer's mailbox in a tight loop. - A telemetry check using Process.info(pid, :message_queue_len) andProcess.info(pid, :memory) to observe the queue size. Create 01-unbounded-mailbox.exs : defmodule SlowConsumer do def start(parent) do spawn(fn -> loop(parent, 0) end) end defp loop(parent, processed_count) do receive do {:item, _item_id} -> # Simulate slow processing (e.g., database query or external API call) Process.sleep(15) updated_count = processed_count + 1 # Periodic logging every 20 items if rem(updated_count, 20) == 0 do {:message_queue_len, queue_len} = Process.info(self(), :message_queue_len) {:memory, memory_bytes} = Process.info(self(), :memory) IO.puts( "Consumer: processed #{updated_count} items | " <> "Mailbox queue length: #{queue_len} | Memory: #{memory_bytes} bytes" ) end loop(parent, updated_count) :stop -> IO.puts("Consumer: stopped") end end end defmodule FastProducer do def produce(consumer_pid, count) do IO.puts("Producer: Blasting #{count} items into consumer mailbox without waiting...") Enum.each(1..count, fn i -> send(consumer_pid, {:item, i}) end) IO.puts("Producer: Finished sending all #{count} items!") end end # ============================================================ # RUN THE EXPERIMENT # ============================================================ IO.puts("=== EXPERIMENT 1: UNBOUNDED MAILBOX GROWTH ===") consumer = SlowConsumer.start(self()) total_items = 100 FastProducer.produce(consumer, total_items) # Process.info/2 inspects the target process externally from the runtime # WITHOUT placing an inspection message in the target's mailbox! {:message_queue_len, queue_len} = Process.info(consumer, :message_queue_len) {:memory, memory} = Process.info(consumer, :memory) IO.puts("\n--- Immediate Snapshot After Producer Finished ---") IO.puts("Items queued in mailbox : #{queue_len}") IO.puts("Process memory usage : #{memory} bytes") IO.puts("--------------------------------------------------\n") # Allow consumer to process remaining backlog IO.puts("Waiting for consumer to drain backlog...") Process.sleep(1700) {:message_queue_len, final_queue_len} = Process.info(consumer, :message_queue_len) {:memory, final_memory} = Process.info(consumer, :memory) IO.puts("\n--- Final Snapshot After Backlog Drained ---") IO.puts("Items queued in mailbox : #{final_queue_len}") IO.puts("Process memory usage : #{final_memory} bytes") IO.puts("---------------------------------------------\n") send(consumer, :stop) Process.sleep(50) IO.puts("Experiment 1 complete.") Running Experiment 1 Execute the script: elixir 01-unbounded-mailbox.exs Output: === EXPERIMENT 1: UNBOUNDED MAILBOX GROWTH === Producer: Blasting 100 items into consumer mailbox without waiting... Producer: Finished sending all 100 items! --- Immediate Snapshot After Producer Finished --- Items queued in mailbox : 99 Process memory usage : 12008 bytes -------------------------------------------------- Waiting for consumer to drain backlog... Consumer: processed 20 items | Mailbox queue length: 80 | Memory: 10360 bytes Consumer: processed 40 items | Mailbox queue length: 60 | Memory: 9104 bytes Consumer: processed 60 items | Mailbox queue length: 40 | Memory: 12280 bytes Consumer: processed 80 items | Mailbox queue length: 20 | Memory: 10576 bytes Consumer: processed 100 items | Mailbox queue length: 0 | Memory: 8648 bytes --- Final Snapshot After Backlog Drained --- Items queued in mailbox : 0 Process memory usage : 8704 bytes --------------------------------------------- Consumer: stopped Experiment 1 complete. What This Tells Us Notice the immediate snapshot: Items queued in mailbox : 99 The producer finished in less than 1 millisecond. But because each item takes 15ms to process, 99 items sat unhandled in the mailbox. If the producer had produced 1,000,000 items instead of 100, the consumer's memory footprint would balloon into gigabytes. Experiment 2: The Lockstep Solution (Stop-and-Wait) How can we prevent the producer from getting ahead of the consumer? The simplest solution is Stop-and-Wait flow control (credit of 1): - The producer sends an item tagged with a unique correlation reference ( make_ref() ). - The producer blocks in receive waiting for an acknowledgment ({:ack, ^ref} ). - The consumer processes the item, then sends {:ack, ref} back to the producer. Producer Consumer | | |--- {:item, self(), ref, 1} ----------------------------->| | | [processes item 1] | | | | [processes item 2] | loop(0) end) end defp loop(processed_count) do receive do {:item, reply_to, ref, item_id} -> # Inspect mailbox before processing: at most 0 other messages waiting! {:message_queue_len, queue_len} = Process.info(self(), :message_queue_len) # Simulate work Process.sleep(15) updated_count = processed_count + 1 if rem(updated_count, 10) == 0 do IO.puts( "Consumer: processed item #{item_id} (total: #{updated_count}) | " <> "Mailbox queue length: #{queue_len}" ) end # Send acknowledgment back to producer send(reply_to, {:ack, ref}) loop(updated_count) :stop -> IO.puts("Consumer: stopped") end end end defmodule AckProducer do def produce(consumer_pid, count) do IO.puts("Producer: Sending #{count} items with stop-and-wait acknowledgment...") Enum.each(1..count, fn i -> ref = make_ref() send(consumer_pid, {:item, self(), ref, i}) # Block until consumer acknowledges receipt of this item receive do {:ack, ^ref} -> :ok after 5000 -> IO.puts("Producer: ERROR - consumer timed out!") end end) IO.puts("Producer: All #{count} items acknowledged by consumer!") end end # ============================================================ # RUN THE EXPERIMENT # ============================================================ IO.puts("=== EXPERIMENT 2: ACK-BASED (STOP-AND-WAIT) FLOW CONTROL ===") consumer = AckConsumer.start() total_items = 50 AckProducer.produce(consumer, total_items) {:message_queue_len, final_queue_len} = Process.info(consumer, :message_queue_len) IO.puts("\nFinal consumer mailbox queue length: #{final_queue_len}") send(consumer, :stop) IO.puts("Experiment 2 complete.") Running Experiment 2 elixir 02-ack-backpressure.exs Output: === EXPERIMENT 2: ACK-BASED (STOP-AND-WAIT) FLOW CONTROL === Producer: Sending 50 items with stop-and-wait acknowledgment... Consumer: processed item 10 (total: 10) | Mailbox queue length: 0 Consumer: processed item 20 (total: 20) | Mailbox queue length: 0 Consumer: processed item 30 (total: 30) | Mailbox queue length: 0 Consumer: processed item 40 (total: 40) | Mailbox queue length: 0 Consumer: processed item 50 (total: 50) | Mailbox queue length: 0 Producer: All 50 items acknowledged by consumer! Final consumer mailbox queue length: 0 Experiment 2 complete. Consumer: stopped The Trade-off of Stop-and-Wait Notice that