Elixir Task.async_stream: Bounded Concurrency, Fault Isolation, and Back-Pressure

The Problem: Unbounded Fan-Out and Uncontrolled Failures

When you need to apply the same operation to thousands of items—API calls, database writes, or file conversions—the naive approach is to spawn a process per item. Without a cap, you exhaust database connections, file handles, or memory. Worse, if one task raises, the default link behavior can take down the caller before it handles the error.

Use Task.Supervisor with Task.async_stream/4 to process an enumerable with a hard ceiling on parallelism, supervised process ownership, and per-item fault isolation.

Desired Outcome

Process a large enumerable concurrently while:

• Capping active tasks at :max_concurrency so downstream systems are not overwhelmed.
• Isolating failures so one bad item produces an {:exit, reason} entry instead of crashing the consumer.
• Keeping every spawned task owned by a supervisor for visibility and cleanup.
• Applying back-pressure so the producer does not outrun the consumer.

Prerequisites

• A running Task.Supervisor in your application's supervision tree.
• An enumerable of work items and a function to apply to each item.
• This guide assumes Elixir 1.15+ and OTP 26+ for option names and defaults. Because Task.async_stream options and defaults have changed across releases, verify the exact supported options for your project's versions.

Procedure

1. Start the Task.Supervisor

Add the supervisor to your application's child list so it starts before any stream runs. If the supervisor is not running, Task.async_stream raises ArgumentError.

# lib/my_app/application.ex
def start(_type, _args) do
  children = [
    {Task.Supervisor, name: MyApp.TaskSupervisor}
  ]

  Supervisor.start_link(children, strategy: :one_for_one)
end

Run this in your normal application boot. No special permissions are required beyond starting the VM.

2. Run the Bounded Stream

Call Task.async_stream/4 with the supervisor as the first argument. The stream yields {:ok, result} for successful items and {:exit, reason} for tasks that terminated abnormally.

# Run in an IEx session attached to your app, or inside a GenServer/module function
user_ids = [1, 2, 3, 4, 5]

results =
  MyApp.TaskSupervisor
  |> Task.async_stream(user_ids, fn id ->
    # Replace with your actual work: HTTP call, DB query, file write
    :timer.sleep(100)
    {:processed, id}
  end,
    max_concurrency: 3,
    ordered: true,
    timeout: 30_000,
    on_timeout: :kill_task
  )
  |> Enum.to_list()

Expected check: results is a list of {:ok, {:processed, id}} tuples in the same order as user_ids because ordered: true. If a task raises, you receive {:exit, reason} at that position instead.

3. Tune Concurrency to Downstream Capacity

Set :max_concurrency to match the bottleneck resource, not the number of CPU cores. If your function uses an Ecto repo with a pool size of 10, set :max_concurrency to 10 or lower. Setting it too high exhausts connections; too low leaves capacity idle.

Option	Effect	When to use
ordered: true	Results arrive in input order	When sequence matters (reports, ordered writes)
ordered: false	Results arrive as tasks finish	When latency matters and order is irrelevant
timeout: N	Kills a task after N ms	When downstream calls can hang
on_timeout: :kill_task	Terminates the task process on timeout	With :timeout to prevent runaway work

4. Isolate and Handle Failures

Because tasks are linked to the caller, an unhandled exit can propagate. The stream captures exits into {:exit, reason} tuples, but you must still handle them explicitly.

results
|> Enum.reduce(%{ok: 0, failed: 0}, fn
  {:ok, _}, acc -> %{acc | ok: acc.ok + 1}
  {:exit, _}, acc -> %{acc | failed: acc.failed + 1}
end)

If you need retries, wrap the consumer in a retry loop or use Stream.transform to re-enqueue failed items. Do not retry indefinitely without a cap.

Expected Checks and Diagnostics

Verify the behavior in an IEx session before deploying.

Confirm the Concurrency Ceiling

Start a supervisor and run a stream where each task sleeps. Count how many are active at once.

# In IEx (iex -S mix)
{:ok, sup} = Task.Supervisor.start_link(name: TestSup)

Task.async_stream(TestSup, 1..10, fn i ->
  :timer.sleep(500)
  i
end, max_concurrency: 3)
|> Stream.each(fn _ ->
  TestSup |> Supervisor.which_children() |> length() |> IO.inspect(label: "active tasks")
end)
|> Enum.to_list()

Expected check: The supervisor tracks at most 3 task processes while the stream is active. The exact count may be lower between batches.

Confirm Fault Isolation

Write a quick test or IEx snippet where the function raises.

Task.async_stream(TestSup, [1, 2, 3], fn
  2 -> raise "boom"
  n -> n
end, max_concurrency: 2)
|> Enum.to_list()

Expected check: The result contains {:exit, %RuntimeError{}} for the failing item and {:ok, 1} / {:ok, 3} for the others. The IEx process does not crash.

Benchmark Against Sequential Processing

Compare Enum.map with Task.async_stream on a representative dataset. Measure wall-clock time at max_concurrency values of 1, 5, 10, and 20. If throughput plateaus, your bottleneck is downstream, not the task layer.

Recovery Options

• Runaway tasks: Always set :timeout and :on_timeout: :kill_task for external calls. A hung HTTP request without a timeout will hold a slot forever.
• Consumer crashes: If the process consuming the stream dies, the linked tasks are terminated. Restart the consumer through its own supervisor; the Task.Supervisor does not restart the consumer.
• Supervisor state corruption: If Supervisor.which_children/1 returns unexpected entries or the supervisor itself crashes, restart it via your application supervisor. This is a state-changing operation—ensure upstream callers can tolerate the interruption.

Limitations

• Task.async_stream links tasks to the caller. If the caller is killed, all in-flight tasks are killed. It is not a fire-and-forget mechanism.
• Option names and defaults have changed across Elixir releases. Verify the exact supported options for your project's Elixir and OTP versions before relying on :on_timeout or :ordered behavior.
• Back-pressure is cooperative: the stream pauses when the consumer is slow, but if the consumer never reads, the producer eventually blocks rather than growing memory unbounded.