Workflows compose jobs together with arbitrary dependencies, allowing sequential, fan-out, and fan-in execution workflows. Workflows are fault tolerant and scale horizontally across all available nodes.

Usage

Workflows linking jobs together into a DAG (directed acyclic graph). Dependency resolution guarantees that jobs execute in the prescribed order regardless of scheduling or retries. Each workflow job will wait for all upstream dependencies to complete before it is made available to run.

As a trivial example, let's define an EchoWorker that only inspects the args, and then use it in a workflow to show how jobs execute in order. First, here's the worker:

defmodule MyApp.EchoWorker do
  use Oban.Pro.Worker, queue: :default

  @impl true
  def process(%{args: args}) do
    IO.inspect(args)

    :ok
  end
end

Now use new/1 to initialize a workflow, and add/4 to add named jobs with dependencies to the workflow:

alias MyApp.EchoWorker
alias Oban.Pro.Workflow

Workflow.new()
|> Workflow.add(:a, EchoWorker.new(%{id: 1}))
|> Workflow.add(:b, EchoWorker.new(%{id: 2}), deps: [:a])
|> Workflow.add(:c, EchoWorker.new(%{id: 3}), deps: [:b])
|> Workflow.add(:d, EchoWorker.new(%{id: 4}), deps: [:b])
|> Workflow.add(:e, EchoWorker.new(%{id: 5}), deps: [:c, :d])
|> Oban.insert_all()

When the workflow executes, it will print out each job's args in the prescribed order. However, because steps c and d each depend on b, they may execute in parallel.

Visually, the workflow jobs composes like this:

Dynamic Workflows

Many workflows aren't static—the number of jobs and their interdependencies aren't known beforehand.

The following worker accepts a count and generates a workflow that fans-out and back in twice, using a variable number of dependencies. The key is using Enum.reduce to accumulate a workflow with interpolated names, i.e. "a_0", "a_1", etc.

defmodule MyApp.Dynamic do
  use Oban.Pro.Worker

  alias Oban.Pro.Workflow

  @impl true
  def process(%{meta: %{"name" => name}}) do
    IO.puts(name)
  end

  def insert_workflow(count) when is_integer(count) do
    range = Range.new(0, count)
    a_deps = Enum.map(range, &"a_#{&1}")
    b_deps = Enum.map(range, &"b_#{&1}")

    Workflow.new()
    |> Workflow.add(:a, new(%{}), [])
    |> fan_out(:a, range)
    |> Workflow.add(:b, new(%{}), deps: a_deps)
    |> fan_out(:b, range)
    |> Workflow.add(:c, new(%{}), deps: b_deps)
    |> Oban.insert_all()
  end

  defp fan_out(workflow, base, range) do
    Enum.reduce(range, workflow, fn key, acc ->
      Workflow.add(acc, "#{base}_#{key}", new(%{}), deps: [base])
    end)
  end
end

Calling MyApp.Dynamic.insert_workflow(3) generates a workflow that fans out to 3 a and 3 b jobs:

Using Upstream Results

Directed dependencies between jobs, paired with the recorded option, allow a workflow's downstream jobs to fetch the output of upstream jobs.

To demonstrate, let's make a workflow that leverages get_recorded/3 to simulate a multi-step API interaction.

The first worker simulates fetching an authentication token using an api_key:

defmodule MyApp.WorkerA do
  use Oban.Pro.Workers.Workflow, recorded: true

  @impl true
  def process(%Job{args: %{"api_key" => api_key}}) do
    token =
      api_key
      |> String.graphemes()
      |> Enum.shuffle()
      |> to_string()

    {:ok, token}
  end
end

The second worker fetches the token from the first job by calling get_job/2 with the name :a, which we'll set while building the workflow later.

defmodule MyApp.WorkerB do
  use Oban.Pro.Worker, recorded: true

  @impl true
  def process(%Job{args: %{"url" => url}} = job) do
    token = Oban.Pro.Workflow.get_recorded(job, :a)

    {:ok, {token, url}}
  end
end

Then the final worker uses all_recorded/3 with the only_deps option to fetch the results from all upstream jobs, then it prints out everything that was fetched.

defmodule MyApp.WorkerC do
  use Oban.Pro.Worker

  @impl true
  def process(job) do
    job
    |> Oban.Pro.Workflow.all_recorded(only_deps: true)
    |> IO.inspect()

    :ok
  end
end

The final step is to build a workflow that composes all of the jobs together with names, args, and deps:

alias MyApp.{WorkerA, WorkerB, WorkerC}

Workflow.new()
|> Workflow.add(:a, WorkerA.new(%{api_key: "23kl239bjljlk309af"}))
|> Workflow.add(:b, WorkerB.new(%{url: "elixir-lang.org"}), deps: [:a])
|> Workflow.add(:c, WorkerB.new(%{url: "www.erlang.org"}), deps: [:a])
|> Workflow.add(:d, WorkerB.new(%{url: "oban.pro"}), deps: [:a])
|> Workflow.add(:e, WorkerC.new(%{}), deps: [:b, :c, :d])
|> Oban.insert_all()

When the workflow runs the final step, e, prints out something like the following:

%{
  "a" => {"93l2jlj3kl90baf2k3", "elixir-lang.org"},
  "b" => {"93l2jlj3kl90baf2k3", "www.erlang.org"},
  "c" => {"93l2jlj3kl90baf2k3", "oban.pro"}
}

Customizing Workflows

Workflow ID

The default workflow_id is a time-ordered, random UUIDv7. This is more than sufficient to ensure that workflows are unique for any period of time. However, if you require more control you can pass a value directly to new/1.

Workflow.new(workflow_id: "custom-but-still-unique-id")

Workflow Name

Workflows accept an optional name to describe the purpose of the workflow, beyond the individual jobs in it. While the workflow_id must be unique, the workflow_name doesn't, so it can be used as a general purpose label.

Workflow.new(workflow_name: "nightly-etl")

Dependency Handling

Workflows use conservative defaults for dependency handling. You can customize the safety checks by providing a few top-level options:

• ignore_cancelled — regard cancelled dependencies as completed rather than cancelling remaining jobs in the workflow. Defaults to false.

• ignore_discarded — regard discarded dependencies as completed rather than cancelling remaining jobs in the workflow. Defaults to false.

• ignore_deleted — regard deleted (typically pruned) dependencies as completed rather cancelling remaining jobs in workflow. Defaults to false.

The following example creates a workflow with all of the available options:

Workflow.new(ignore_cancelled: true, ignore_deleted: true, ignore_discarded: true)

Options may also be applied to individual workflow jobs For example, configure a single job to ignore cancelled dependencies, another to ignore discarded, and another to ignore deleted:

Workflow.new()
|> Workflow.add(:a, MyWorkflow.new(%{}))
|> Workflow.add(:b, MyWorkflow.new(%{}, deps: [:a], ignore_cancelled: true))
|> Workflow.add(:c, MyWorkflow.new(%{}, deps: [:b], ignore_discarded: true))
|> Workflow.add(:d, MyWorkflow.new(%{}, deps: [:c], ignore_deleted: true))

Stuck Workflows

Be sure that you're running the DynamicLifeline to rescue stuck workflows when upstream dependencies are deleted unexpectedly.

 config :my_app, Oban,
   plugins: [Oban.Pro.Plugins.DynamicLifeline],
   ...

Fetching Workflow Jobs

Workflow jobs are tied together through meta attributes. The get_job/3, all_jobs/3, and stream_jobs/3 functions use those attributes to load other jobs in a workflow. This is particularly useful from a worker's process/1 function. For example, to fetch all of the jobs in a workflow:

defmodule MyApp.Workflow do
  use Oban.Pro.Worker

  @impl true
  def process(%Job{} = job) do
    job
    |> Oban.Pro.Workflow.all_jobs()
    |> do_things_with_jobs()

    :ok
  end
end

It's also possible to scope fetching to only dependencies of the current job with only_deps:

deps = Workflow.all_jobs(job, only_deps: true)

Or, only fetch a single explicit dependency by name get_job/3:

dep_job = Workflow.get_job(job, :a)

For large workflows it may be inefficient to load all jobs in memory at once. In that case, you can use stream_jobs/3 to fetch jobs lazily. For example, to stream all of the completed jobs in a workflow:

defmodule MyApp.Workflow do
  use Oban.Pro.Worker

  @impl true
  def process(%Job{} = job) do
    {:ok, workflow_jobs} =
      MyApp.Repo.transaction(fn ->
        job
        |> Oban.Pro.Workflow.stream_jobs()
        |> Stream.filter(& &1.state == "completed")
        |> Enum.to_list()
      end)

    do_things_with_jobs(workflow_jobs)

    :ok
  end
end

Streaming is provided by Ecto's Repo.stream, and it must take place within a transaction. Using a stream lets you control the number of jobs loaded from the database, minimizing memory usage for large workflows.

Appending Workflow Jobs

Sometimes all jobs aren't known when the workflow is created. In that case, you can add more jobs with optional dependency checking using append/2. An appended workflow starts with one or more jobs, which reuses the original workflow_id, and optionally builds a set of dependencies to check against.

In this example we disable deps checking with check_deps: false:

defmodule MyApp.WorkflowWorker do
  use Oban.Pro.Worker

  alias Oban.Pro.Workflow

  @impl true
  def process(%Job{} = job) do
    jobs =
      job
      |> Workflow.append(check_deps: false)
      |> Workflow.add(:d, WorkerD.new(%{}), deps: [:a])
      |> Workflow.add(:e, WorkerE.new(%{}), deps: [:b])
      |> Workflow.add(:f, WorkerF.new(%{}), deps: [:c])
      |> Oban.insert_all()

    {:ok, jobs}
  end
end

The new jobs specify deps on preexisting jobs named :a, :b, and :c, but there isn't any guarantee those jobs actually exist. That could lead to an incomplete workflow where the new jobs may never complete.

To be safe and check jobs while appending we'll fetch all of the previous jobs with all_jobs/3 and feed them in:

defmodule MyApp.WorkflowWorker do
  use Oban.Pro.Worker

  alias Oban.Pro.Workflow

  @impl true
  def process(%Job{} = job) do
    {:ok, jobs} = Workflow.all_jobs(job)

    jobs
    |> Workflow.append()
    |> Workflow.add(:d, WorkerD.new(%{}), deps: [:a])
    |> Workflow.add(:e, WorkerE.new(%{}), deps: [:b])
    |> Workflow.add(:f, WorkerF.new(%{}), deps: [:c])
    |> Oban.insert_all()

    :ok
  end
end

Now there isn't any risk of an incomplete workflow from missing dependencies, at the expense of loading some extraneous jobs.

Handling Cancellations

Workflow jobs are automatically cancelled when their upstream dependencies are cancelled, discarded, or deleted (unless specifically overridden using the ignore_* options as described earlier). Those workflow jobs are cancelled before they're executing, which means standard Oban.Pro.Worker.after_process/3 hooks won't be called. Instead, there's an optional after_cancelled/2 callback specifically for workflows.

Here's a trivial after_cancelled hook that logs a warning when a workflow job is cancelled:

def MyApp.Workflow do
  use Oban.Pro.Worker

  @behaviour Oban.Pro.Workflow

  require Logger

  @impl Oban.Pro.Workflow
  def after_cancelled(reason, job) do
    Logger.warn("Workflow job #{job.id} cancelled because a dependency was #{reason}")
  end

Visualizing Workflows

Workflows are a type of Directed Acyclic Graph, which means we can describe a workflow as a graph of jobs and dependencies, where execution flows between jobs. By converting the workflow into a standard graph description language such as mermaid or graphviz, we can visualize workflows.

Along with the general purpose to_graph/1 that outputs a digraph, there are a few built-in visualization options: to_dot/1 for graphviz digraphs and to_mermaid/1 for mermaid flowcharts.

The following example generates a mermaid flowchart from the account workflow from above:

Workflow.to_mermaid(archive_account_workflow(123))

Generates the following mermaid output, where each vertex is the job's name and the label is the worker:

flowchart TD
    backup_1[MyApp.BackupPost] --> delete[MyApp.DeleteAccount]
    backup_2[MyApp.BackupPost] --> delete[MyApp.DeleteAccount]
    backup_3[MyApp.BackupPost] --> delete[MyApp.DeleteAccount]
    receipt[MyApp.FinalReceipt] --> backup_1[MyApp.BackupPost]
    receipt[MyApp.FinalReceipt] --> backup_2[MyApp.BackupPost]
    receipt[MyApp.FinalReceipt] --> backup_3[MyApp.BackupPost]
    receipt[MyApp.FinalReceipt] --> email_1[MyApp.EmailSubscriber]
    receipt[MyApp.FinalReceipt] --> email_2[MyApp.EmailSubscriber]
    email_1[MyApp.EmailSubscriber] --> delete[MyApp.DeleteAccount]
    email_2[MyApp.EmailSubscriber] --> delete[MyApp.DeleteAccount]

Now we can take that output and render it using the command line, or directly in LiveBook using Kino. The following example pipes a workflow through to_mermaid/1 and then into Kino.Mermaid:

workflow
|> Workflow.to_mermaid()
|> Kino.Mermaid.new()

That will generate the following SVG of the workflow:

Looking at the flowchart we can clearly see how the workflow starts with a single render job, fans-out to multiple email and backup jobs, and finally fans-in to the delete job—exactly as we planned!