3122 lines
92 KiB
Elixir
3122 lines
92 KiB
Elixir
defmodule Oban.Pro.Workflow do
|
|
@moduledoc ~S"""
|
|
Workflows compose jobs with flexible dependency relationships, enabling sequential execution,
|
|
fan-out parallelization, and fan-in convergence patterns. By declaratively defining jobs and
|
|
their connections, you can build complex processing pipelines that are both fault-tolerant and
|
|
horizontally scalable across all nodes.
|
|
|
|
## Basic Usage
|
|
|
|
Workflows are composed of regular jobs or decorated functions linked together through
|
|
declarative dependencies. A workflow consists of these key elements:
|
|
|
|
- **Jobs** - Individual units of work identified by unique names
|
|
- **Dependencies** - Relationships specifying execution order
|
|
|
|
Let's create a workflow using a decorated `echo/1` function that demonstrates how jobs execute
|
|
in order:
|
|
|
|
```elixir
|
|
defmodule EchoWorkflow do
|
|
use Oban.Pro.Decorator
|
|
|
|
alias Oban.Pro.Workflow
|
|
|
|
def insert do
|
|
Workflow.new()
|
|
|> Workflow.add(:a, new_echo(1))
|
|
|> Workflow.add(:b, new_echo(2), deps: :a)
|
|
|> Workflow.add(:c, new_echo(3), deps: :b)
|
|
|> Workflow.add(:d, new_echo(4), deps: :b)
|
|
|> Workflow.add(:e, new_echo(5), deps: [:c, :d])
|
|
|> Oban.insert_all()
|
|
end
|
|
|
|
@job true
|
|
def echo(value), do: IO.inspect(value)
|
|
end
|
|
```
|
|
|
|
The workflow is initialized with `Workflow.new/1`, jobs are added with `Workflow.add/3`,
|
|
connections between jobs are added with the `deps` option, and finally the workflow is inserted
|
|
with `Oban.insert_all/1`.
|
|
|
|
When executed, this workflow will print each job's `value` in the prescribed order. Steps `:c`
|
|
and `:d` both depend on `:b`, so they may execute in parallel. Visually, the workflow looks like
|
|
this:
|
|
|
|
|
|
|
|
This short example already covers the shapes most workflows reach for: sequential execution
|
|
(`:a → :b`), fan-out where one job triggers several (`:b → :c`, `:b → :d`), and fan-in where
|
|
multiple upstreams converge (`:c`, `:d → :e`).
|
|
|
|
## Sub-Workflows
|
|
|
|
Workflows can be nested hierarchically using sub workflows. This allows you to compose complex
|
|
workflows from simpler ones, making it easier to organize and reuse workflow patterns.
|
|
|
|
The `add_workflow/4` function lets you add an entire workflow as a dependency of another
|
|
workflow. Like `add/4`, it accepts a name and optional dependencies, but instead of a job
|
|
changeset, it takes another workflow.
|
|
|
|
Here's an example of creating a primary workflow and adding sub workflows with dependencies:
|
|
|
|
```elixir
|
|
alias MyApp.{WorkerA, WorkerB, WorkerC}
|
|
alias Oban.Pro.Workflow
|
|
|
|
extr_flow =
|
|
Workflow.new(name: "extract")
|
|
|> Workflow.add(:extract, WorkerA.new(%{source: "database"}))
|
|
|> Workflow.add(:transform, WorkerB.new(%{type: "normalize"}), deps: :extract)
|
|
|
|
note_flow =
|
|
Workflow.new(name: "notify")
|
|
|> Workflow.add(:prepare, WorkerA.new(%{template: "report"}))
|
|
|> Workflow.add(:send, WorkerB.new(%{method: "email"}), deps: :prepare)
|
|
|
|
# Create the main workflow and add sub workflows to it
|
|
Workflow.new()
|
|
|> Workflow.add(:setup, WorkerC.new(%{mode: "initialize"}))
|
|
|> Workflow.add_workflow(:extract, extr_flow, deps: :setup)
|
|
|> Workflow.add_workflow(:notify, note_flow, deps: :extract)
|
|
|> Workflow.add(:finalize, WorkerC.new(%{mode: "cleanup"}), deps: :notify)
|
|
|> Oban.insert_all()
|
|
```
|
|
|
|
In this example, the main workflow has a single job named `:setup`, followed by two sub
|
|
workflows, and ends with a `:finalize` job. The dependencies ensure proper execution order:
|
|
`setup -> extraction -> notification -> finalize`.
|
|
|
|
Sub-workflows are a powerful pattern for:
|
|
|
|
- Organizing large job dependencies into logical units
|
|
- Enabling reuse across applications
|
|
- Simplifying the maintenance of complex job graphs
|
|
|
|
## Sharing Results
|
|
|
|
Directed dependencies between jobs, paired with the `recorded` option, allow downstream jobs to
|
|
fetch the output of upstream jobs. This is particularly useful for multi-step processes where
|
|
each step builds on previous results.
|
|
|
|
Consider this workflow that simulates a multi-step API interaction:
|
|
|
|
```elixir
|
|
defmodule MyApp.WorkerA do
|
|
use Oban.Pro.Worker, recorded: true
|
|
|
|
@impl true
|
|
def process(%Job{args: %{"api_key" => api_key}}) do
|
|
token =
|
|
api_key
|
|
|> String.graphemes()
|
|
|> Enum.shuffle()
|
|
|> to_string()
|
|
|
|
{:ok, token} # This return value will be recorded
|
|
end
|
|
end
|
|
```
|
|
|
|
The second worker fetches the `token` from the first job by calling `get_recorded/2` with the name
|
|
`:a`, which we'll set while building the workflow later.
|
|
|
|
```elixir
|
|
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.
|
|
|
|
```elixir
|
|
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
|
|
```
|
|
|
|
Now compose the workers together:
|
|
|
|
```elixir
|
|
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 prints something like:
|
|
|
|
```elixir
|
|
%{
|
|
"b" => {"93l2jlj3kl90baf2k3", "elixir-lang.org"},
|
|
"c" => {"93l2jlj3kl90baf2k3", "www.erlang.org"},
|
|
"d" => {"93l2jlj3kl90baf2k3", "oban.pro"}
|
|
}
|
|
```
|
|
|
|
Sharing results between jobs is a powerful building block for processing pipelines. While the
|
|
approach above works well, cascading functions provide an even more elegant way to share and
|
|
transform data between workflow steps.
|
|
|
|
## Cascading Functions
|
|
|
|
Cascade mode allows you to build workflows using function captures that automatically receive
|
|
context and previous step results. Each function receives a map containing the workflow context
|
|
and the results of its dependencies. Results from each step are recorded and made available to
|
|
subsequent steps.
|
|
|
|
Here's an ETL (Extract, Transform, Load) pipeline using cascading functions:
|
|
|
|
```elixir
|
|
defmodule MyApp.ETL do
|
|
def insert(source, date) do
|
|
Workflow.new()
|
|
|> Workflow.put_context(%{source: source, date: date})
|
|
|> Workflow.add_cascade(:extract, &extract/1)
|
|
|> Workflow.add_cascade(:transform, &transform/1, deps: :extract)
|
|
|> Workflow.add_cascade(:load, &load/1, deps: :transform)
|
|
|> Workflow.add_cascade(:notify, ¬ify/1, deps: [:transform, :load])
|
|
|> Oban.insert_all()
|
|
end
|
|
|
|
def extract(%{source: source, date: date}) do
|
|
if source =~ ~r/[a-z0-9]+/ do
|
|
%{records: [1, 2, 3], extracted_at: DateTime.utc_now()}
|
|
else
|
|
{:cancel, "unprocessable source"}
|
|
end
|
|
end
|
|
|
|
def transform(%{extract: %{records: records}}) do
|
|
transformed = Enum.map(records, & &1 * 2)
|
|
%{records: transformed, count: length(transformed)}
|
|
end
|
|
|
|
def load(%{transform: %{records: records}}) do
|
|
%{loaded: true, count: length(records)}
|
|
end
|
|
|
|
def notify(%{transform: transform, load: load}) do
|
|
IO.puts("Processed #{transform.count} records, loaded #{load.count}")
|
|
end
|
|
end
|
|
```
|
|
|
|
When this workflow runs, each function is automatically called with a context map containing the
|
|
shared context (source and date), and results from dependencies (accessible by their step
|
|
names).
|
|
|
|
There's no need to manually fetch recorded results as with standard workflow jobs! For standard
|
|
workflows that need to fetch results from dependencies, including grafted sub-workflows, see
|
|
`all_recorded/3`.
|
|
|
|
Cascading functions are especially valuable for:
|
|
|
|
- Data pipelines where each step needs the output of previous steps
|
|
- Validating aggregate context before proceeding
|
|
- Minimizing worker and function definition boilerplate
|
|
|
|
### Cascading Sub-Workflow
|
|
|
|
Cascades can be added as sub-workflows that efficiently fan out workloads across multiple items.
|
|
This works similar to `add_many/4`, but for cascading functions.
|
|
|
|
To create a fan-out cascade, provide a tuple containing an enumerable (like a list, map, or
|
|
range) and a function capture with arity 2. The function will be called once for each item in
|
|
the enumerable, with the first argument receiving the current item and the second argument
|
|
receiving the cumulative context.
|
|
|
|
```elixir
|
|
defmodule MyApp.BatchETL do
|
|
def insert(sources, date) do
|
|
Workflow.new()
|
|
|> Workflow.put_context(%{date: date})
|
|
|> Workflow.add_cascade(:sources, {sources, &extract_source/2})
|
|
|> Workflow.add_cascade(:transform, &transform_all/1, deps: :sources)
|
|
|> Workflow.add_cascade(:load, &load_data/1, deps: :transform)
|
|
|> Workflow.add_cascade(:notify, &send_notification/1, deps: [:transform, :load])
|
|
|> Oban.insert_all()
|
|
end
|
|
|
|
# Each `source` from the list is passed as the first argument
|
|
def extract_source(source, context) do
|
|
%{source: source, records: [1, 2, 3], extracted_at: DateTime.utc_now()}
|
|
end
|
|
|
|
def transform_all(%{sources: sources}) do
|
|
total_records = Enum.sum_by(sources, fn {_key, result} -> length(result.records) end)
|
|
|
|
transformed =
|
|
sources
|
|
|> Enum.flat_map(fn {_key, %{records: records}} -> records end)
|
|
|> Enum.map(&(&1 * 2))
|
|
|
|
%{records: transformed, count: total_records}
|
|
end
|
|
|
|
def load_data(%{transform: %{records: records}}) do
|
|
%{loaded: true, count: length(records)}
|
|
end
|
|
|
|
def send_notification(%{transform: transform, load: load, date: date}) do
|
|
IO.puts("On #{date}, processed #{transform.count} records, loaded #{load.count}")
|
|
end
|
|
end
|
|
```
|
|
|
|
This example processes multiple data sources in parallel, then consolidates the results in
|
|
subsequent steps. The `extract_source/2` function is automatically called for each source in the
|
|
provided list, with the results collected into a map.
|
|
|
|
Cascading sub-workflows are useful for:
|
|
|
|
- Processing collections of items in parallel with shared context
|
|
- Building data pipelines that operate on multiple sources simultaneously
|
|
- Implementing fan-out/fan-in patterns with minimal boilerplate
|
|
- Maintaining clean separation between item processing and result aggregation
|
|
|
|
## Dynamic Workflows
|
|
|
|
Many workflows aren't static—the number of jobs and their interdependencies aren't known
|
|
beforehand. You can generate workflows dynamically based on runtime conditions.
|
|
|
|
The following worker creates a workflow that fans-out and back in twice, using a variable number
|
|
of dependencies:
|
|
|
|
```elixir
|
|
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
|
|
```
|
|
|
|
This approach is useful for:
|
|
|
|
- Processing variable-sized collections
|
|
- Creating workflows based on database queries
|
|
- Building complex workflows from configuration
|
|
- Implementing multi-tenant workflows with different requirements
|
|
|
|
### Grafting Sub-Workflows
|
|
|
|
Grafting allows sub-workflows to be attached after a workflow has started. With grafting, you
|
|
define placeholders that are expanded into full sub-workflows later when data becomes available.
|
|
|
|
Graft jobs serve as placeholders in a workflow. When a graft job executes, it _must_ build
|
|
and attach a new sub-workflow at that point. Any downstream jobs that depend on the grafter
|
|
will wait for the entire grafted sub-workflow to complete.
|
|
|
|
Here's an example workflow that takes an `account_id` then grafts on a sub-workflow to contact
|
|
all users in the account:
|
|
|
|
```elixir
|
|
defmodule MyApp.AccountNotifier do
|
|
alias Oban.Pro.Workflow
|
|
|
|
def insert(account_id) do
|
|
Workflow.new()
|
|
|> Workflow.put_context(%{account_id: account_id})
|
|
|> Workflow.add_graft(:users, &graft_users/1)
|
|
|> Workflow.add_cascade(:notify, &send_summary/1, deps: :users)
|
|
|> Oban.insert_all()
|
|
end
|
|
|
|
def graft_users(%{account_id: account_id}) do
|
|
user_ids = MyApp.Account.get_user_ids!(account_id)
|
|
|
|
{user_ids, &send_notice/2}
|
|
|> Workflow.apply_graft()
|
|
|> Oban.insert_all()
|
|
end
|
|
|
|
def send_notice(user_id, context) do
|
|
# Notify users
|
|
end
|
|
|
|
def send_summary(%{account_id: account_id, users: user_ids}) do
|
|
# Deliver summary
|
|
end
|
|
end
|
|
```
|
|
|
|
The summary job waits for the grafting job and the grafted workflow to complete before
|
|
executing.
|
|
|
|
This pattern is particularly useful for:
|
|
|
|
- Building sub-workflows dynamically based on data discovered during execution
|
|
- Conditionally executing different workflow branches
|
|
- Managing resource-intensive sub-workflows
|
|
|
|
### Appending 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`:
|
|
|
|
```elixir
|
|
def process(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
|
|
```
|
|
|
|
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:
|
|
|
|
```elixir
|
|
def process(job) do
|
|
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
|
|
```
|
|
|
|
Now there isn't any risk of an incomplete workflow from missing dependencies, at the expense of
|
|
loading some extraneous jobs.
|
|
|
|
Appending jobs is particularly useful for:
|
|
|
|
- Extending workflows based on user interaction or external events
|
|
- Adding additional reporting or cleanup steps after seeing initial results
|
|
- Breaking large workflows into stages that are built incrementally
|
|
|
|
## Awaiting Signals
|
|
|
|
Workflow jobs can pause mid-execution to wait for an external decision, then resume with the
|
|
signal payload. This turns workflows into durable state machines that can wait for human
|
|
approval, third-party callbacks, or any other out-of-band event without holding a process or
|
|
database connection open.
|
|
|
|
Pause a job with `Oban.Pro.Worker.await_signal/1` and resume it with `signal/3,4` from anywhere—
|
|
another job, a Phoenix controller, an IEx session, etc:
|
|
|
|
```elixir
|
|
defmodule MyApp.Approval do
|
|
use Oban.Pro.Worker
|
|
|
|
@impl Oban.Pro.Worker
|
|
def process(_job) do
|
|
case Oban.Pro.Worker.await_signal(wait_for: {24, :hours}) do
|
|
{:ok, %{decision: "approved"}} -> charge_card()
|
|
{:ok, %{decision: "rejected"}} -> {:cancel, :rejected}
|
|
{:error, :timeout} -> {:cancel, :no_decision}
|
|
end
|
|
end
|
|
end
|
|
```
|
|
|
|
Then signal the job by its workflow name when the decision arrives:
|
|
|
|
```elixir
|
|
Workflow.signal(workflow_id, :approval, %{decision: "approved"})
|
|
```
|
|
|
|
A short live wait (`:wait_timeout`, default 5s) blocks during execution so signals that arrive
|
|
immediately resume in place. Past that window the job snoozes until either a signal arrives or
|
|
the overall `:wait_for` deadline elapses, freeing the worker process during the wait. Signals are
|
|
persisted, so a signal delivered before `await_signal/1` is reached is consumed on the next
|
|
call.
|
|
|
|
See `Oban.Pro.Worker.await_signal/1` for the full set of options.
|
|
|
|
## Unique Workflows
|
|
|
|
Pass `unique: true` with a `name` to ensure only one workflow with that name is active at a time.
|
|
While a uniquely named workflow is running, attempts to insert another with the same name are
|
|
marked as conflicts and their jobs aren't inserted, much like Oban's unique jobs.
|
|
|
|
Workflow.new(unique: true, name: "nightly-etl")
|
|
|> Workflow.add(:extract, Extract.new(%{}))
|
|
|> Workflow.add(:load, Load.new(%{}), deps: :extract)
|
|
|> Oban.insert_all()
|
|
|
|
Uniqueness is keyed on the workflow's `name`, so a `name` is required. Creating a unique workflow
|
|
without one raises an `ArgumentError`.
|
|
|
|
Conflicts are detected at insert time with row locks, which makes them safe across concurrent
|
|
nodes. When a workflow conflicts none of its jobs are inserted, and the returned jobs are flagged
|
|
with `conflict?: true`:
|
|
|
|
# While a "nightly-etl" workflow is still running...
|
|
|
|
[job] =
|
|
Workflow.new(unique: true, name: "nightly-etl")
|
|
|> Workflow.add(:extract, Extract.new(%{}))
|
|
|> Oban.insert_all()
|
|
|
|
job.conflict? #=> true
|
|
|
|
A name is only considered taken while the workflow is incomplete. Once every job finishes and the
|
|
workflow reaches a `completed` state, the name is free and a new workflow may reuse it.
|
|
|
|
## Customization
|
|
|
|
Workflows provide several customization options to control their behavior, from identification
|
|
to error handling. This section covers how to tailor workflows to your specific requirements.
|
|
|
|
### Workflow ID
|
|
|
|
Every workflow needs a unique identifier. By default, `workflow_id` is a time-ordered, random
|
|
[UUIDv7][uuid], which ensures uniqueness for any period. For more control, you can provide a
|
|
custom ID:
|
|
|
|
```elixir
|
|
Workflow.new(workflow_id: "custom-but-still-unique-id")
|
|
```
|
|
|
|
[uuid]: https://datatracker.ietf.org/doc/html/draft-peabody-dispatch-new-uuid-format-04#section-5.2
|
|
|
|
### Workflow Name
|
|
|
|
Additionally, workflows accept an optional name that describes their purpose:
|
|
|
|
```elixir
|
|
Workflow.new(name: "nightly-etl")
|
|
```
|
|
|
|
While the `workflow_id` must be unique, the `name` doesn't have to be, so it can serve
|
|
as a general-purpose label for categorizing or grouping workflows.
|
|
|
|
### Context Key Format
|
|
|
|
By default, cascade functions receive context with atom keys for top level dependencies.
|
|
However, _nested_ dependencies will have string keys. To prevent this inconsistency you can opt
|
|
in to atomizing string keys to atoms with the `atom_keys` option:
|
|
|
|
* `atom_keys` — Atomize top level keys for cascade context. Defaults to `true`.
|
|
|
|
Apply this option when creating a workflow:
|
|
|
|
```elixir
|
|
Workflow.new(atom_keys: true)
|
|
|> Workflow.add_cascade(:fetch, &MyApp.fetch/1)
|
|
|> Workflow.add_cascade(:process, &MyApp.process/1, deps: :fetch)
|
|
```
|
|
|
|
With `atom_keys: false`, the `process/1` function receives `%{"fetch" => result}` instead of
|
|
`%{fetch: result}`.
|
|
|
|
### Dependency Handling
|
|
|
|
By default, workflows use conservative dependency handling - if an upstream job is cancelled,
|
|
discarded, or deleted, dependent jobs are automatically cancelled. You can customize this
|
|
behavior with these options:
|
|
|
|
* `ignore_cancelled` — Treat `cancelled` dependencies as completed rather than cancelling
|
|
downstream jobs. Defaults to `false`.
|
|
|
|
* `ignore_discarded` — Treat `discarded` dependencies as completed rather than cancelling
|
|
downstream jobs. Defaults to `false`.
|
|
|
|
* `ignore_deleted` — Treat `deleted` (typically pruned) dependencies as completed rather than
|
|
cancelling downstream jobs. Defaults to `false`.
|
|
|
|
Apply these options to the entire workflow:
|
|
|
|
```elixir
|
|
Workflow.new(ignore_cancelled: true, ignore_deleted: true, ignore_discarded: true)
|
|
```
|
|
|
|
Or configure individual jobs within a workflow:
|
|
|
|
```elixir
|
|
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)
|
|
```
|
|
|
|
> #### Preventing Stuck Workflows {: .tip}
|
|
>
|
|
> Be sure that you're running the `DynamicLifeline` to rescue stuck workflows when upstream
|
|
> dependencies are deleted unexpectedly.
|
|
>
|
|
> ```elixir
|
|
> config :my_app, Oban,
|
|
> plugins: [Oban.Pro.Plugins.DynamicLifeline],
|
|
> ...
|
|
> ```
|
|
|
|
### Cancellation Callbacks
|
|
|
|
Workflow jobs are automatically `cancelled` when their upstream dependencies are `cancelled`,
|
|
`discarded`, or `deleted` (unless specifically configured with the `ignore_*` options mentioned
|
|
above). Since these workflow jobs are cancelled before execution, standard
|
|
`c:Oban.Pro.Worker.after_process/3` hooks won't be called.
|
|
|
|
Instead, you can implement the optional `c:after_cancelled/2` callback specifically for
|
|
workflows:
|
|
|
|
```elixir
|
|
defmodule 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}")
|
|
|
|
# Perform any cleanup required
|
|
notify_monitoring_system(job, reason)
|
|
|
|
:ok
|
|
end
|
|
end
|
|
```
|
|
|
|
The `reason` parameter will be one of:
|
|
|
|
- `:cancelled` - A dependency was cancelled
|
|
- `:discarded` - A dependency was discarded
|
|
- `:deleted` - A dependency was deleted
|
|
|
|
### Handling Failures with Status Checking
|
|
|
|
For situations where you can't use the `after_cancelled` callback, or when you have many
|
|
different workers in a workflow, you can combine the `ignore_cancelled` option with status
|
|
checking in a downstream job.
|
|
|
|
This pattern allows a final "cleanup" job to examine the state of all upstream jobs and take
|
|
appropriate action:
|
|
|
|
```elixir
|
|
Workflow.new(ignore_cancelled: true)
|
|
|> Workflow.add(:step_1, StepOne.new(%{}))
|
|
|> Workflow.add(:step_2, StepTwo.new(%{}), deps: :step_1)
|
|
|> Workflow.add(:step_3, StepThree.new(%{}), deps: :step_2)
|
|
|> Workflow.add(:report, Reporter.new(%{}), deps: ~w(step_1 step_2 step_3))
|
|
```
|
|
|
|
Then, in your final worker, examine the state of the upstream jobs using `status/1`:
|
|
|
|
```elixir
|
|
defmodule MyApp.Reporter do
|
|
use Oban.Pro.Worker
|
|
|
|
@impl true
|
|
def process(job) do
|
|
status = Workflow.status(job)
|
|
|
|
if status.counts.cancelled > 0 do
|
|
# Some results were cancelled
|
|
else
|
|
:ok
|
|
end
|
|
end
|
|
end
|
|
```
|
|
|
|
This approach offers several advantages over hooks:
|
|
|
|
- It works with any type of worker, including decorated functions and cascades
|
|
- You can centralize error handling in a single place
|
|
- The workflow continues to the reporting step even if some steps fail
|
|
|
|
## Introspection
|
|
|
|
Workflow jobs are tied together through `meta` attributes. You can retrieve these jobs using
|
|
several functions, each with different optimization characteristics.
|
|
|
|
### Fetching Jobs
|
|
|
|
There are several functions for retrieving jobs, `get_job/2` and `all_jobs/3`, which are
|
|
optimized for different use cases:
|
|
|
|
```elixir
|
|
# Get a single specific job by name
|
|
job_a = Workflow.get_job(job, :a)
|
|
|
|
# Get all jobs in a workflow
|
|
all_jobs = Workflow.all_jobs(job)
|
|
|
|
# Get only the dependencies of the current job
|
|
deps = Workflow.all_jobs(job, only_deps: true)
|
|
|
|
# Get specific named jobs
|
|
[job_a, job_b] = Workflow.all_jobs(job, names: [:a, :b])
|
|
|
|
# Access from outside a worker using workflow_id
|
|
jobs = Workflow.all_jobs("some-uuid-1234-5678")
|
|
```
|
|
|
|
### Streaming Jobs
|
|
|
|
For large workflows where loading all jobs at once would consume too much memory, use
|
|
`stream_jobs/3`:
|
|
|
|
```elixir
|
|
def process(job) do
|
|
{:ok, workflow_jobs} =
|
|
MyApp.Repo.transaction(fn ->
|
|
job
|
|
|> Oban.Pro.Workflow.stream_jobs()
|
|
|> Stream.filter(& &1.state == "completed")
|
|
|> Enum.to_list()
|
|
end)
|
|
|
|
process_completed_jobs(workflow_jobs)
|
|
|
|
:ok
|
|
end
|
|
```
|
|
|
|
> #### Transaction Required {: .info}
|
|
>
|
|
> Streaming is provided by Ecto's `Repo.stream`, and must take place within a transaction.
|
|
> This approach lets you control the number of jobs loaded from the database, minimizing memory
|
|
> usage for large workflows.
|
|
|
|
### Retrieving Results
|
|
|
|
To fetch recorded output from workflow jobs, without loading the full job first, use
|
|
`get_recorded/2` and `all_recorded/3`:
|
|
|
|
```elixir
|
|
# Get results from a specific job
|
|
result_a = Workflow.get_recorded(job, :a)
|
|
|
|
# Get results from all jobs
|
|
results = Workflow.all_recorded(job)
|
|
|
|
# Get only results from dependencies
|
|
dep_results = Workflow.all_recorded(job, only_deps: true)
|
|
```
|
|
|
|
### Checking Status
|
|
|
|
To get detailed status information about a workflow, including the overall workflow state, job
|
|
counts, time measurement, and sub-workflow details, use `status/1`:
|
|
|
|
```elixir
|
|
status = Workflow.status(job)
|
|
```
|
|
|
|
This outputs a map with details, somewhat like this:
|
|
|
|
```elixir
|
|
%{
|
|
id: "some-uuid-1234-5678",
|
|
name: "my-workflow",
|
|
total: 3,
|
|
state: :completed,
|
|
duration: 72,
|
|
counts: %{
|
|
cancelled: 0,
|
|
available: 0,
|
|
completed: 3,
|
|
scheduled: 0,
|
|
discarded: 0,
|
|
executing: 0,
|
|
retryable: 0
|
|
},
|
|
started_at: ~U[2025-04-10 16:34:34],
|
|
stopped_at: ~U[2025-04-10 16:34:34],
|
|
subs: %{}
|
|
}
|
|
```
|
|
|
|
## Visualization
|
|
|
|
Workflows are a type of a [Directed Acyclic Graph][dag], which means we can represent them
|
|
visually as nodes (jobs) connected by edges (dependencies). By converting workflows into
|
|
standard graph description languages, we can create clear visual representations of job
|
|
execution patterns.
|
|
|
|
### Converting to Graph Formats
|
|
|
|
Along with the general purpose `to_graph/1` that outputs a `digraph`, there are several built-in
|
|
visualization options:
|
|
|
|
* `to_dot/1` - Generates [GraphViz][gv] DOT format
|
|
* `to_mermaid/1` - Produces [Mermaid][mermaid] flowchart markup
|
|
|
|
For example, let's generate a mermaid flowchart from our account workflow:
|
|
|
|
```elixir
|
|
Workflow.to_mermaid(archive_account_workflow(123))
|
|
```
|
|
|
|
This produces the following mermaid output, where each vertex represents a job's name and the
|
|
label shows the worker:
|
|
|
|
```text
|
|
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]
|
|
```
|
|
|
|
### Rendering Visualizations
|
|
|
|
You can render these visualizations using the command line with a `mermaid` binary, directly in
|
|
[LiveBook][live] using [Kino][kino], or in documentation sites.
|
|
|
|
For LiveBook, you can pipe a workflow through `to_mermaid/1` and then into `Kino.Mermaid`:
|
|
|
|
```elixir
|
|
workflow
|
|
|> Workflow.to_mermaid()
|
|
|> Kino.Mermaid.new()
|
|
```
|
|
|
|
This generates an SVG diagram like this:
|
|
|
|
|
|
|
|
The visualization clearly shows how our workflow starts with a single `receipt` job, fans-out to
|
|
multiple `email` and `backup` jobs, and finally fans-in to the `delete` job—making complex
|
|
dependency relationships immediately apparent.
|
|
|
|
[dag]: https://en.wikipedia.org/wiki/Directed_acyclic_graph
|
|
[mermaid]: https://mermaid.js.org/
|
|
[gv]: https://graphviz.org
|
|
[live]: https://livebook.dev/
|
|
[kino]: https://hexdocs.pm/kino/Kino.Mermaid.html
|
|
"""
|
|
|
|
@behaviour Oban.Pro.Flusher
|
|
@behaviour Oban.Pro.Handler
|
|
|
|
import Ecto.Query
|
|
|
|
alias Ecto.Changeset
|
|
alias Oban.{Config, Job, Repo, Validation}
|
|
alias Oban.Pro.Stages.{Chain, Hooks}
|
|
alias Oban.Pro.Utils
|
|
alias Oban.Pro.Workflow.{Cascade, Schema}
|
|
|
|
# Types
|
|
|
|
@type add_opt ::
|
|
{:deps, name() | [name()]}
|
|
| {:ignore_cancelled, boolean()}
|
|
| {:ignore_deleted, boolean()}
|
|
| {:ignore_discarded, boolean()}
|
|
|
|
@type add_opts :: [add_opt()]
|
|
|
|
@type add_cascade_opts :: [Job.option() | add_opt()]
|
|
|
|
@type add_workflow_opts :: [deps: name() | [name()]]
|
|
|
|
@type append_opts :: [
|
|
check_deps: boolean(),
|
|
ignore_cancelled: boolean(),
|
|
ignore_deleted: boolean(),
|
|
ignore_discarded: boolean(),
|
|
workflow_id: String.t(),
|
|
workflow_name: String.t()
|
|
]
|
|
|
|
@type cancel_opts :: [names: [name()]]
|
|
|
|
@type cancel_reason :: :deleted | :discarded | :cancelled
|
|
|
|
@type cascade_capture :: (map() -> any()) | {Enum.t(), (any(), map() -> any())}
|
|
|
|
@type fetch_opts :: [
|
|
log: Logger.level(),
|
|
names: [name()],
|
|
only_deps: boolean(),
|
|
timeout: timeout(),
|
|
with_subs: boolean()
|
|
]
|
|
|
|
@type job_or_wid :: Job.t() | workflow_id()
|
|
|
|
@type name :: atom() | String.t()
|
|
|
|
@type new_opts :: [
|
|
id: String.t(),
|
|
ignore_cancelled: boolean(),
|
|
ignore_deleted: boolean(),
|
|
ignore_discarded: boolean(),
|
|
name: String.t(),
|
|
unique: boolean(),
|
|
workflow_id: String.t(),
|
|
workflow_name: String.t()
|
|
]
|
|
|
|
@type status :: %{
|
|
id: workflow_id(),
|
|
name: String.t() | nil,
|
|
total: non_neg_integer(),
|
|
state: :executing | :completed | :cancelled | :discarded | :unknown,
|
|
counts: %{atom() => non_neg_integer()},
|
|
duration: non_neg_integer() | nil,
|
|
started_at: DateTime.t() | nil,
|
|
stopped_at: DateTime.t() | nil,
|
|
subs: %{String.t() => map()}
|
|
}
|
|
|
|
@type t :: %__MODULE__{
|
|
id: workflow_id(),
|
|
changesets: [Job.changeset()],
|
|
check_deps: boolean(),
|
|
names: MapSet.t(),
|
|
opts: map()
|
|
}
|
|
|
|
@type workflow_id :: String.t()
|
|
|
|
# Callbacks
|
|
|
|
@doc deprecated: "Use the universal `c:Oban.Pro.Worker.on_cancelled/2` callback instead"
|
|
@callback after_cancelled(cancel_reason(), job :: Job.t()) :: :ok
|
|
|
|
@optional_callbacks after_cancelled: 2
|
|
|
|
# Constants
|
|
|
|
@all_states Enum.map(Job.states(), &to_string/1)
|
|
@incomplete_set MapSet.new(~w(suspended available executing retryable scheduled))
|
|
@chan_keys ~w(max_attempts meta priority queue schedule_in scheduled_at tags)a
|
|
@work_keys ~w(ignore_cancelled ignore_deleted ignore_discarded workflow_id workflow_name)a
|
|
|
|
# Macros
|
|
|
|
defstruct [
|
|
:id,
|
|
changesets: [],
|
|
check_deps: true,
|
|
eliminations: %{},
|
|
grafts: %{},
|
|
names: MapSet.new(),
|
|
opts: %{},
|
|
subs: %{}
|
|
]
|
|
|
|
defguardp is_job_or_wid(job_or_wid) when is_struct(job_or_wid, Job) or is_binary(job_or_wid)
|
|
|
|
defguardp is_name(name) when (is_atom(name) and not is_nil(name)) or is_binary(name)
|
|
|
|
# This is a replacement for `push`, which uses `array_append` and isn't compatible with jsonb
|
|
# arrays. The `||` operator works with both arrays and jsonb.
|
|
defmacrop concat_errors(column, error) do
|
|
quote do
|
|
fragment("? || ?", unquote(column), unquote(error))
|
|
end
|
|
end
|
|
|
|
defmacrop drop_hold(meta) do
|
|
quote do
|
|
fragment(~s(? || '{"on_hold":false}'), unquote(meta))
|
|
end
|
|
end
|
|
|
|
defmacrop in_workflow(meta, workflow_id) do
|
|
quote do
|
|
fragment(
|
|
"? \\? 'workflow_id' AND ?->>'workflow_id' = ?",
|
|
unquote(meta),
|
|
unquote(meta),
|
|
unquote(workflow_id)
|
|
)
|
|
end
|
|
end
|
|
|
|
defmacrop in_workflow_or_sup(meta, workflow_id) do
|
|
quote do
|
|
fragment(
|
|
"((? \\? 'workflow_id' AND ?->>'workflow_id' = ?) OR (? \\? 'sup_workflow_id' AND ?->>'sup_workflow_id' = ?))",
|
|
unquote(meta),
|
|
unquote(meta),
|
|
unquote(workflow_id),
|
|
unquote(meta),
|
|
unquote(meta),
|
|
unquote(workflow_id)
|
|
)
|
|
end
|
|
end
|
|
|
|
defmacrop in_suspended(meta, workflow_id) do
|
|
quote do
|
|
fragment(
|
|
"? \\? 'workflow_id' AND ?->>'workflow_id' = ?",
|
|
unquote(meta),
|
|
unquote(meta),
|
|
unquote(workflow_id)
|
|
)
|
|
end
|
|
end
|
|
|
|
defmacrop in_suspended_sup(meta, workflow_id) do
|
|
quote do
|
|
fragment(
|
|
"? \\? 'sup_workflow_id' AND ?->>'sup_workflow_id' = ?",
|
|
unquote(meta),
|
|
unquote(meta),
|
|
unquote(workflow_id)
|
|
)
|
|
end
|
|
end
|
|
|
|
defmacrop in_legacy_suspended(meta, workflow_id) do
|
|
quote do
|
|
fragment(
|
|
"? \\? 'workflow_id' AND ?->>'workflow_id' = ? AND ?->>'on_hold' = 'true'",
|
|
unquote(meta),
|
|
unquote(meta),
|
|
unquote(workflow_id),
|
|
unquote(meta)
|
|
)
|
|
end
|
|
end
|
|
|
|
defmacrop in_legacy_suspended_sup(meta, workflow_id) do
|
|
quote do
|
|
fragment(
|
|
"? \\? 'sup_workflow_id' AND ?->>'sup_workflow_id' = ? AND ?->>'on_hold' = 'true'",
|
|
unquote(meta),
|
|
unquote(meta),
|
|
unquote(workflow_id),
|
|
unquote(meta)
|
|
)
|
|
end
|
|
end
|
|
|
|
# Telemetry
|
|
|
|
@impl Oban.Pro.Handler
|
|
def on_start do
|
|
events = [
|
|
[:oban, :engine, :cancel_all_jobs, :stop],
|
|
[:oban, :plugin, :stop]
|
|
]
|
|
|
|
:telemetry.attach_many("oban.pro.workflow", events, &__MODULE__.handle_event/4, nil)
|
|
end
|
|
|
|
@impl Oban.Pro.Handler
|
|
def on_stop do
|
|
:telemetry.detach("oban.pro.workflow")
|
|
end
|
|
|
|
@doc false
|
|
def handle_event([:oban, :plugin, _], _time, %{conf: conf, discarded_jobs: jobs}, nil) do
|
|
for %{meta: %{"workflow_id" => workflow_id}} <- jobs do
|
|
on_flush(workflow_id, conf)
|
|
end
|
|
end
|
|
|
|
def handle_event([:oban, :engine, _, _], _time, %{conf: conf, jobs: jobs}, nil) do
|
|
for %{meta: %{"workflow_id" => workflow_id}} <- jobs do
|
|
on_flush(workflow_id, conf)
|
|
end
|
|
end
|
|
|
|
def handle_event(_event, _time, _meta, _conf), do: :ok
|
|
|
|
# Handling
|
|
|
|
@doc false
|
|
@impl Oban.Pro.Flusher
|
|
def to_flush_mfa(job, conf) do
|
|
case job.meta do
|
|
%{"ancestor_ids" => ancestors, "workflow_id" => workflow_id} ->
|
|
{__MODULE__, :on_flush, [Enum.uniq([workflow_id | ancestors]), conf]}
|
|
|
|
%{"sup_workflow_id" => sup_workflow_id, "workflow_id" => workflow_id} ->
|
|
{__MODULE__, :on_flush, [[sup_workflow_id, workflow_id], conf]}
|
|
|
|
%{"workflow_id" => workflow_id} ->
|
|
{__MODULE__, :on_flush, [[workflow_id], conf]}
|
|
|
|
_ ->
|
|
:ignore
|
|
end
|
|
end
|
|
|
|
# CTE macros for dependency state lookup query
|
|
# These extract prefix/wid once and build clean SQL fragments
|
|
|
|
# Extract values once from held to avoid repeated TOAST access
|
|
defmacrop held_prep_cte do
|
|
quote do
|
|
fragment("""
|
|
SELECT h.id, h.meta, h.meta->>'workflow_id' AS job_wid, h.meta->'deps' AS deps
|
|
FROM "held" h
|
|
""")
|
|
end
|
|
end
|
|
|
|
defmacrop workflow_jobs_cte(prefix, wid) do
|
|
quote do
|
|
fragment(
|
|
"""
|
|
SELECT meta->>'name' AS name, state
|
|
FROM ?.oban_jobs
|
|
WHERE meta \\? 'workflow_id' AND meta->>'workflow_id' = ?
|
|
""",
|
|
identifier(^unquote(prefix)),
|
|
^unquote(wid)
|
|
)
|
|
end
|
|
end
|
|
|
|
defmacrop job_deps_cte do
|
|
quote do
|
|
fragment("""
|
|
SELECT h.id AS job_id, h.job_wid, dep, jsonb_typeof(dep) AS dep_type
|
|
FROM held_prep h, jsonb_array_elements(h.deps) AS dep
|
|
""")
|
|
end
|
|
end
|
|
|
|
defmacrop dep_lookup_cte(prefix, wid) do
|
|
quote do
|
|
fragment(
|
|
"""
|
|
SELECT jd.job_id, wj.state
|
|
FROM job_deps jd
|
|
LEFT JOIN "workflow_jobs" wj ON wj.name = jd.dep #>> '{}'
|
|
WHERE jd.dep_type = 'string' AND jd.job_wid = ?
|
|
|
|
UNION ALL
|
|
|
|
SELECT jd.job_id, oj.state
|
|
FROM "job_deps" jd
|
|
LEFT JOIN ?.oban_jobs oj ON
|
|
oj.meta \\? 'workflow_id'
|
|
AND oj.meta->>'workflow_id' = jd.job_wid
|
|
AND oj.meta->>'name' = jd.dep #>> '{}'
|
|
WHERE jd.dep_type = 'string' AND jd.job_wid <> ?
|
|
|
|
UNION ALL
|
|
|
|
SELECT jd.job_id, oj.state
|
|
FROM "job_deps" jd
|
|
LEFT JOIN ?.oban_jobs oj ON
|
|
oj.meta \\? 'workflow_id'
|
|
AND oj.meta->>'workflow_id' = jd.dep->>0
|
|
AND oj.meta->>'name' = jd.dep->>1
|
|
WHERE jd.dep_type = 'array' AND jd.dep->>1 <> '*'
|
|
|
|
UNION ALL
|
|
|
|
SELECT jd.job_id, oj.state
|
|
FROM "job_deps" jd
|
|
LEFT JOIN ?.oban_jobs oj ON
|
|
(oj.meta \\? 'workflow_id' AND oj.meta->>'workflow_id' = jd.dep->>0)
|
|
OR (oj.meta \\? 'sup_workflow_id' AND oj.meta->>'sup_workflow_id' = jd.dep->>0)
|
|
WHERE jd.dep_type = 'array' AND jd.dep->>1 = '*'
|
|
""",
|
|
^unquote(wid),
|
|
identifier(^unquote(prefix)),
|
|
^unquote(wid),
|
|
identifier(^unquote(prefix)),
|
|
identifier(^unquote(prefix))
|
|
)
|
|
end
|
|
end
|
|
|
|
defmacrop job_states_cte do
|
|
quote do
|
|
fragment("""
|
|
SELECT job_id, coalesce(array_agg(state::text) FILTER (WHERE state IS NOT NULL), ARRAY[]::text[]) AS states
|
|
FROM dep_lookup
|
|
GROUP BY job_id
|
|
""")
|
|
end
|
|
end
|
|
|
|
@doc false
|
|
def on_flush(wid, conf) when is_binary(wid) do
|
|
held_query = lock(Job, "FOR UPDATE SKIP LOCKED")
|
|
|
|
held_query =
|
|
if Utils.has_legacy_workflows?(conf) do
|
|
where(
|
|
held_query,
|
|
[j],
|
|
(j.state == "suspended" and in_suspended(j.meta, ^wid)) or
|
|
(j.state == "suspended" and in_suspended_sup(j.meta, ^wid)) or
|
|
(j.state == "scheduled" and in_legacy_suspended(j.meta, ^wid)) or
|
|
(j.state == "scheduled" and in_legacy_suspended_sup(j.meta, ^wid))
|
|
)
|
|
else
|
|
held_query
|
|
|> where([j], j.state == "suspended")
|
|
|> where([j], in_suspended(j.meta, ^wid) or in_suspended_sup(j.meta, ^wid))
|
|
end
|
|
|
|
prefix = conf.prefix
|
|
|
|
query =
|
|
Job
|
|
|> with_cte("held", as: ^held_query)
|
|
|> with_cte("held_prep", as: held_prep_cte(), materialized: true)
|
|
|> with_cte("workflow_jobs", as: workflow_jobs_cte(prefix, wid), materialized: true)
|
|
|> with_cte("job_deps", as: job_deps_cte(), materialized: true)
|
|
|> with_cte("dep_lookup", as: dep_lookup_cte(prefix, wid))
|
|
|> with_cte("job_states", as: job_states_cte())
|
|
|> join(:inner, [j], h in fragment(~s("held")), on: j.id == h.id)
|
|
|> join(:left, [j, h], js in fragment(~s("job_states")), on: js.job_id == h.id)
|
|
|> select([j, h, js], {j.id, j.meta, js.states})
|
|
|
|
conf
|
|
|> Repo.all(query)
|
|
|> Enum.map(&to_stage_operation/1)
|
|
|> Enum.group_by(&elem(&1, 0), &elem(&1, 1))
|
|
|> Enum.map(&apply_operations(&1, conf))
|
|
|> on_flush(conf)
|
|
end
|
|
|
|
def on_flush(workflow_ids, conf) when is_list(workflow_ids) do
|
|
workflow_ids
|
|
|> List.flatten()
|
|
|> Enum.each(&on_flush(&1, conf))
|
|
end
|
|
|
|
@doc """
|
|
Instantiate a new workflow struct with a unique workflow id.
|
|
|
|
## Examples
|
|
|
|
Create a standard workflow without any options:
|
|
|
|
Workflow.new()
|
|
|
|
Create a workflow with a custom name:
|
|
|
|
Workflow.new(workflow_name: "logistics")
|
|
|
|
Create a workflow with a static id and some options:
|
|
|
|
Workflow.new(workflow_id: "workflow-id", ignore_cancelled: true, ignore_discarded: true)
|
|
"""
|
|
@spec new(opts :: new_opts()) :: t()
|
|
def new(opts \\ []) do
|
|
opts =
|
|
opts
|
|
|> expand_alias(:id, :workflow_id)
|
|
|> expand_alias(:name, :workflow_name)
|
|
|
|
validate!(opts)
|
|
|
|
opts =
|
|
opts
|
|
|> Keyword.put(:workflow, true)
|
|
|> Keyword.put_new_lazy(:workflow_id, &Oban.Pro.UUIDv7.generate/0)
|
|
|> Map.new()
|
|
|
|
%__MODULE__{id: opts.workflow_id, opts: opts}
|
|
end
|
|
|
|
defp expand_alias(opts, short, full) do
|
|
case Keyword.pop(opts, short) do
|
|
{nil, opts} -> opts
|
|
{value, opts} -> Keyword.put_new(opts, full, value)
|
|
end
|
|
end
|
|
|
|
defp validate!(opts) do
|
|
Validation.validate_schema!(opts,
|
|
atom_keys: :boolean,
|
|
ignore_cancelled: :boolean,
|
|
ignore_deleted: :boolean,
|
|
ignore_discarded: :boolean,
|
|
unique: :boolean,
|
|
workflow_id: :string,
|
|
workflow_name: :string
|
|
)
|
|
|
|
if opts[:unique] == true and is_nil(opts[:workflow_name]) do
|
|
raise ArgumentError, "the :name option is required when :unique is true"
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Add a named job to the workflow along with optional dependencies.
|
|
|
|
## Examples
|
|
|
|
Add jobs to a workflow with dependencies:
|
|
|
|
Workflow.new()
|
|
|> Workflow.add(:a, WorkerA.new(%{id: id}))
|
|
|> Workflow.add(:b, WorkerB.new(%{id: id}), deps: :a)
|
|
|> Workflow.add(:c, WorkerC.new(%{id: id}), deps: :a)
|
|
|> Workflow.add(:d, WorkerC.new(%{id: id}), deps: [:b, :c])
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec add(t(), name(), Job.changeset(), add_opts()) :: t()
|
|
def add(%__MODULE__{} = workflow, name, changeset, opts \\ []) when is_name(name) do
|
|
{deps, opts} = Keyword.pop(opts, :deps, [])
|
|
|
|
deps = expand_deps(deps, workflow)
|
|
name = to_string(name)
|
|
|
|
prevent_chain!(changeset, name)
|
|
prevent_dupe!(workflow, name)
|
|
confirm_deps!(workflow, deps)
|
|
|
|
meta =
|
|
changeset
|
|
|> Changeset.get_change(:meta, %{})
|
|
|> Map.put(:deps, deps)
|
|
|> Map.put(:name, name)
|
|
|> Map.put(:workflow_id, workflow.id)
|
|
|> Map.merge(workflow.opts)
|
|
|> Map.merge(Map.new(opts))
|
|
|
|
meta =
|
|
if Changeset.get_field(changeset, :state) == "scheduled" do
|
|
orig_at =
|
|
changeset
|
|
|> Changeset.get_change(:scheduled_at)
|
|
|> DateTime.to_unix(:microsecond)
|
|
|
|
Map.put(meta, :orig_scheduled_at, orig_at)
|
|
else
|
|
meta
|
|
end
|
|
|
|
changeset =
|
|
if Enum.any?(deps) do
|
|
changeset
|
|
|> Changeset.put_change(:meta, meta)
|
|
|> Changeset.put_change(:state, "suspended")
|
|
else
|
|
Changeset.put_change(changeset, :meta, meta)
|
|
end
|
|
|
|
changesets = workflow.changesets ++ [changeset]
|
|
|
|
%{workflow | changesets: changesets, names: MapSet.put(workflow.names, name)}
|
|
end
|
|
|
|
defp expand_deps(deps, workflow, mode \\ :main) do
|
|
deps
|
|
|> List.wrap()
|
|
|> Enum.reduce([], fn name, acc ->
|
|
name = to_string(name)
|
|
|
|
cond do
|
|
is_map_key(workflow.grafts, name) ->
|
|
grafter_name = "grafter_#{name}"
|
|
|
|
grafter_dep =
|
|
if is_map_key(workflow.subs, grafter_name) do
|
|
[Map.get(workflow.subs, grafter_name), "*"]
|
|
else
|
|
grafter_name
|
|
end
|
|
|
|
[grafter_dep | [[Map.get(workflow.subs, name), "*"] | acc]]
|
|
|
|
is_map_key(workflow.eliminations, name) ->
|
|
Map.get(workflow.eliminations, name) ++ acc
|
|
|
|
is_map_key(workflow.subs, name) ->
|
|
[[Map.get(workflow.subs, name), "*"] | acc]
|
|
|
|
mode == :sub ->
|
|
[[workflow.id, name] | acc]
|
|
|
|
true ->
|
|
[name | acc]
|
|
end
|
|
end)
|
|
|> Enum.reverse()
|
|
end
|
|
|
|
@doc """
|
|
Add a cascading function to the workflow with optional dependencies.
|
|
|
|
Cascading functions receive a context map containing the workflow's shared context and any
|
|
results from their upstream dependencies. Results of the function are automatically recorded,
|
|
unless they're one of the following standard control tuples:
|
|
|
|
- `{:cancel, reason}`
|
|
- `{:error, reason}`
|
|
- `{:snooze, seconds}`
|
|
|
|
> #### Function Captures Only {: .warning}
|
|
>
|
|
> Cascading functions _must_ be captures like `&MyApp.foo/1` for external functions or `&foo/1`
|
|
> for local functions. Anonymous functions won't work. The function always receives a single
|
|
> map parameter containing all context and previous results.
|
|
|
|
## Cascading Sub-Workflow
|
|
|
|
Cascades can be added as sub-workflows that efficiently fan out workloads across multiple items.
|
|
This works similar to `add_many/4`, but for cascading functions.
|
|
|
|
To create a fan-out cascade, provide a tuple containing an enumerable (like a list, map, or
|
|
range) and a function capture with arity 2. The function will be called once for each item in
|
|
the enumerable, with the first argument receiving the current item and the second argument
|
|
receiving the cumulative context.
|
|
|
|
```elixir
|
|
user_ids = [101, 202, 303]
|
|
|
|
Workflow.new()
|
|
|> Workflow.put_context(%{operation: "sync"})
|
|
|> Workflow.add_cascade(:users, {user_ids, &MyApp.sync_user/2})
|
|
|> Workflow.add_cascade(:notify, &MyApp.send_summary/1, deps: :users)
|
|
|
|
def sync_user(user_id, context) do
|
|
# ...
|
|
end
|
|
```
|
|
|
|
## Options
|
|
|
|
A subset of standard job options can be passed directly to customize the cascade jobs. The
|
|
available options are:
|
|
|
|
* `:max_attempts`
|
|
* `:meta`
|
|
* `:priority`
|
|
* `:queue`
|
|
* `:schedule_in`
|
|
* `:scheduled_at`
|
|
* `:tags`
|
|
|
|
Pass the additional options alongside `deps` or other `add/4` workflow options.
|
|
|
|
## Examples
|
|
|
|
Add a cascading function to a workflow:
|
|
|
|
Workflow.new()
|
|
|> Workflow.put_context(%{user_id: 123})
|
|
|> Workflow.add_cascade(:activate, &MyApp.activate/1)
|
|
|> Workflow.add_cascade(:notify, &MyApp.notify/1, deps: :activate)
|
|
|> Oban.insert_all()
|
|
|
|
Customize the cascade job:
|
|
|
|
Workflow.new()
|
|
|> Workflow.add_cascade(:activate, &MyApp.activate/1, queue: :activation)
|
|
|> Workflow.add_cascade(:retrofit, &MyApp.retrofit/1, max_attempts: 3)
|
|
|
|
Add cascade steps from a range:
|
|
|
|
Workflow.new()
|
|
|> Workflow.put_context(%{user_id: 123})
|
|
|> Workflow.add_cascade(:activations, {1..5, &MyApp.activate/2})
|
|
|
|
Add cascade steps from a map, using the map keys as custom names:
|
|
|
|
accounts = %{admin: 1, user: 2, guest: 3}
|
|
|
|
Workflow.new()
|
|
|> Workflow.put_context(%{action: "sync"})
|
|
|> Workflow.add_cascade(:accounts, {accounts, &MyApp.sync_account/2})
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec add_cascade(t(), name(), cascade_capture(), add_cascade_opts()) :: t()
|
|
def add_cascade(workflow, name, fun, opts \\ [])
|
|
|
|
def add_cascade(%__MODULE__{} = workflow, name, fun, opts) when is_function(fun) do
|
|
{changeset, opts} = build_cascade(fun, opts)
|
|
|
|
add(workflow, name, changeset, opts)
|
|
end
|
|
|
|
def add_cascade(%__MODULE__{} = workflow, name, {items, fun}, opts)
|
|
when is_name(name) and is_function(fun, 2) do
|
|
info = Function.info(fun)
|
|
|
|
cond do
|
|
info[:arity] != 2 ->
|
|
raise ArgumentError, "context function must have arity of 2"
|
|
|
|
info[:module] == :erl_eval ->
|
|
raise ArgumentError, "context function must be a capture and not anonymous"
|
|
|
|
true ->
|
|
argf = &%{arg: &1, mod: inspect(info[:module]), fun: to_string(info[:name])}
|
|
decn = "#{inspect(info[:module])}.#{info[:name]}/#{info[:arity]}"
|
|
meta = %{decorated: true, decorated_name: decn}
|
|
|
|
{chan_opts, opts} = Keyword.split(opts, @chan_keys)
|
|
{work_opts, opts} = Keyword.split(opts, @work_keys)
|
|
|
|
chan_opts = Keyword.update(chan_opts, :meta, meta, &Map.merge(&1, meta))
|
|
|
|
sub =
|
|
items
|
|
|> with_names()
|
|
|> Enum.reduce(new(work_opts), fn {key, item}, acc ->
|
|
add(acc, to_string(key), Cascade.new(argf.(item), chan_opts))
|
|
end)
|
|
|
|
add_workflow(workflow, name, sub, opts)
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Adds a grafting point to the workflow where sub-workflows can be attached later.
|
|
|
|
A grafter serves as a placeholder in the workflow topology that downstream jobs can
|
|
depend upon. Later, another job can dynamically generate a sub-workflow and attach it
|
|
at this grafting point using `graft/3`.
|
|
|
|
The grafter can be defined using either a job changeset or a cascading function.
|
|
|
|
## Examples
|
|
|
|
Add a grafting point with a job changeset:
|
|
|
|
Workflow.new()
|
|
|> Workflow.add_graft(:users, MyWorker.new(%{}))
|
|
|> Workflow.add(:notify, WorkerC.new(%{}), deps: :users)
|
|
|
|
Add a grafting point with a cascading function:
|
|
|
|
Workflow.new()
|
|
|> Workflow.put_context(%{account_id: account_id})
|
|
|> Workflow.add_graft(:users, &build_users_workflow/1)
|
|
|> Workflow.add_cascade(:notify, &send_summary/1, deps: :users)
|
|
|
|
Add multiple grafting points with a fan-out tuple:
|
|
|
|
Workflow.new()
|
|
|> Workflow.add_graft(:accounts, {account_ids, &sync_account/2})
|
|
|> Workflow.add(:finalize, FinalizerWorker.new(%{}), deps: :accounts)
|
|
|
|
Each function in a fan-out graft receives the item and context, and should call
|
|
`apply_graft/1` to dynamically create a sub-workflow:
|
|
|
|
def sync_account(account_id, _context) do
|
|
{user_ids(account_id), &sync_user/2}
|
|
|> Workflow.apply_graft()
|
|
|> Oban.insert_all()
|
|
|
|
:ok
|
|
end
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec add_graft(t(), name(), Job.changeset() | cascade_capture(), add_cascade_opts()) :: t()
|
|
def add_graft(workflow, name, changeset_or_fun, opts \\ [])
|
|
|
|
def add_graft(%__MODULE__{} = workflow, name, fun, opts) when is_function(fun) do
|
|
{changeset, opts} = build_cascade(fun, opts)
|
|
|
|
add_graft(workflow, name, changeset, opts)
|
|
end
|
|
|
|
def add_graft(%__MODULE__{} = workflow, name, {_items, fun} = cascade, opts)
|
|
when is_name(name) and is_function(fun, 2) do
|
|
name = to_string(name)
|
|
uuid = "grafted_#{Oban.Pro.UUIDv7.generate()}"
|
|
grafter_name = "grafter_#{name}"
|
|
|
|
graft_meta = %{graft_id: uuid, graft_name: name}
|
|
opts = Keyword.update(opts, :meta, graft_meta, &Map.merge(&1, graft_meta))
|
|
|
|
workflow = %{
|
|
workflow
|
|
| grafts: Map.put(workflow.grafts, name, grafter_name),
|
|
subs: Map.put(workflow.subs, name, uuid)
|
|
}
|
|
|
|
add_cascade(workflow, grafter_name, cascade, opts)
|
|
end
|
|
|
|
def add_graft(%__MODULE__{} = workflow, name, changeset, opts) when is_name(name) do
|
|
name = to_string(name)
|
|
uuid = "grafted_#{Oban.Pro.UUIDv7.generate()}"
|
|
grafter_name = "grafter_#{name}"
|
|
|
|
meta =
|
|
changeset
|
|
|> Changeset.get_change(:meta, %{})
|
|
|> Map.put(:graft_id, uuid)
|
|
|> Map.put(:graft_name, name)
|
|
|
|
changeset = Changeset.put_change(changeset, :meta, meta)
|
|
|
|
workflow =
|
|
workflow
|
|
|> Map.update!(:grafts, &Map.put(&1, name, grafter_name))
|
|
|> Map.update!(:subs, &Map.put(&1, name, uuid))
|
|
|
|
add(workflow, grafter_name, changeset, opts)
|
|
end
|
|
|
|
defp build_cascade(fun, opts) do
|
|
info = Function.info(fun)
|
|
|
|
cond do
|
|
info[:arity] != 1 ->
|
|
raise ArgumentError, "context function must have arity 1"
|
|
|
|
info[:module] == :erl_eval ->
|
|
raise ArgumentError, "context function must be a capture and not anonymous"
|
|
|
|
true ->
|
|
args = %{mod: inspect(info[:module]), fun: to_string(info[:name])}
|
|
decn = "#{inspect(info[:module])}.#{info[:name]}/#{info[:arity]}"
|
|
meta = %{decorated: true, decorated_name: decn}
|
|
|
|
{chan_opts, opts} =
|
|
opts
|
|
|> Keyword.update(:meta, meta, &Map.merge(&1, meta))
|
|
|> Keyword.split(@chan_keys)
|
|
|
|
{Cascade.new(args, chan_opts), opts}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Add multiple named jobs as a sub workflow along with optional dependencies.
|
|
|
|
## Empty Sub-Workflows
|
|
|
|
When an empty collection of jobs is provided, the sub-workflow step is skipped entirely.
|
|
Any dependencies on this step are automatically transferred to downstream steps to maintain
|
|
the correct execution order.
|
|
|
|
Empty sub-workflows have no job results to retrieve or cascade outputs to reference. Consider
|
|
this when creating dependencies on sub-workflows that might be empty at runtime.
|
|
|
|
## Examples
|
|
|
|
Add multiple jobs as a sub workflow with a list, where the keys will be the job's position,
|
|
starting at 0:
|
|
|
|
Workflow.add_many(workflow, :sub_1, [WorkerA.new(%{}), WorkerB.new(%{})])
|
|
|
|
Add multiple jobs as a sub workflow with a map, where the keys are the job names:
|
|
|
|
Workflow.add_many(workflow, :sub_1, %{a: WorkerA.new(%{}), b: WorkerB.new(%{})})
|
|
|
|
Add multiple jobs as sub workflows and add a dependency on them:
|
|
|
|
Workflow.new()
|
|
|> Workflow.add_many(:a, changesets_a)
|
|
|> Workflow.add_many(:b, changesets_b)
|
|
|> Workflow.add(:c, WorkerC.new(%{}), deps: [:a, :b])
|
|
|
|
Workflow options pay be provided to name or otherwise customize the sub-workflow:
|
|
|
|
Workflow.add_many(workflow, :sub_1, changesets, ignore_cancelled: true)
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec add_many(t(), name(), Enum.t(), new_opts() | add_opts()) :: t()
|
|
def add_many(%__MODULE__{} = workflow, name, changesets, opts \\ []) when is_name(name) do
|
|
{work_opts, opts} = Keyword.split(opts, @work_keys)
|
|
|
|
sub =
|
|
changesets
|
|
|> with_names()
|
|
|> Enum.reduce(new(work_opts), fn {key, chg}, acc -> add(acc, to_string(key), chg) end)
|
|
|
|
add_workflow(workflow, name, sub, opts)
|
|
end
|
|
|
|
defp with_names(items) when is_list(items), do: Enum.with_index(items, &{&2, &1})
|
|
defp with_names(items) when is_struct(items), do: items |> Enum.to_list() |> with_names()
|
|
defp with_names(items) when is_map(items), do: items
|
|
|
|
@doc """
|
|
Add a sub workflow with a name and optional dependencies to another workflow.
|
|
|
|
This function lets you compose workflows by adding one workflow as a dependency within another
|
|
workflow. Workflows may depend on jobs or other workflows, and jobs may depend on entire
|
|
workflows. If a job depends on a workflow, it won't execute until every job in the workflow is
|
|
in a final state (completed, cancelled, or discarded).
|
|
|
|
## Examples
|
|
|
|
Add a sub workflow to a workflow with dependencies:
|
|
|
|
sub_1 =
|
|
Workflow.new()
|
|
|> Workflow.add(:a, WorkerB.new(%{id: id}))
|
|
|> Workflow.add(:b, WorkerC.new(%{id: id}))
|
|
|
|
sub_2 =
|
|
Workflow.new()
|
|
|> Workflow.add(:a, WorkerB.new(%{id: id}))
|
|
|> Workflow.add(:b, WorkerC.new(%{id: id}))
|
|
|
|
Workflow.new()
|
|
|> Workflow.add(:a, WorkerA.new(%{id: id}))
|
|
|> Workflow.add_workflow(:b, sub_1, deps: :a)
|
|
|> Workflow.add_workflow(:c, sub_2, deps: :b)
|
|
|> Workflow.add(:d, WorkerD.new(%{id: id}), deps: [:b, :c])
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec add_workflow(t(), name(), t(), add_workflow_opts()) :: t()
|
|
def add_workflow(%_{} = workflow, name, %_{} = sub_workflow, opts \\ []) when is_name(name) do
|
|
{deps, opts} = Keyword.pop(opts, :deps, [])
|
|
|
|
deps = expand_deps(deps, workflow, :sub)
|
|
name = to_string(name)
|
|
|
|
prevent_dupe!(workflow, name)
|
|
confirm_deps!(workflow, deps)
|
|
|
|
sub_changesets =
|
|
Enum.map(sub_workflow.changesets, fn changeset ->
|
|
meta_deps = get_in(changeset.changes, [:meta, :deps])
|
|
|
|
if deps != [] and meta_deps == [] do
|
|
old_ancestors = get_in(changeset.changes, [:meta, :ancestor_ids]) || []
|
|
new_ancestors = Map.get(workflow.opts, :ancestor_ids, [])
|
|
all_ancestors = old_ancestors ++ [workflow.id] ++ new_ancestors
|
|
|
|
opts
|
|
|> Map.new()
|
|
|> Map.put(:ancestor_ids, all_ancestors)
|
|
|> Map.put(:deps, deps)
|
|
|> Map.put(:sub_name, name)
|
|
|> Map.put(:sup_workflow_id, workflow.id)
|
|
|> Map.put(:workflow_id, sub_workflow.id)
|
|
|> put_hold(changeset)
|
|
else
|
|
new_ancestors = Map.get(workflow.opts, :ancestor_ids, [])
|
|
|
|
meta =
|
|
if get_in(changeset.changes, [:meta, :sup_workflow_id]) do
|
|
old_ancestors = get_in(changeset.changes, [:meta, :ancestor_ids]) || []
|
|
all_ancestors = old_ancestors ++ [workflow.id] ++ new_ancestors
|
|
|
|
%{sup_workflow_id: workflow.id, ancestor_ids: all_ancestors}
|
|
else
|
|
%{
|
|
sub_name: name,
|
|
sup_workflow_id: workflow.id,
|
|
workflow_id: sub_workflow.id,
|
|
ancestor_ids: [workflow.id] ++ new_ancestors
|
|
}
|
|
end
|
|
|
|
Changeset.update_change(changeset, :meta, &Map.merge(&1, meta))
|
|
end
|
|
end)
|
|
|
|
subs = Map.put(workflow.subs, name, sub_workflow.id)
|
|
|
|
case sub_changesets do
|
|
[_ | _] ->
|
|
changesets = workflow.changesets ++ sub_changesets
|
|
|
|
%{workflow | changesets: changesets, subs: subs}
|
|
|
|
[] ->
|
|
eliminations = Map.put(workflow.eliminations, name, deps)
|
|
|
|
%{workflow | eliminations: eliminations, subs: subs}
|
|
end
|
|
end
|
|
|
|
defp put_hold(meta, changeset) do
|
|
changeset
|
|
|> Changeset.update_change(:meta, &Map.merge(&1, meta))
|
|
|> Changeset.put_change(:state, "suspended")
|
|
end
|
|
|
|
@doc """
|
|
Adds a context value to the workflow that can be accessed by all jobs.
|
|
|
|
This is a convenience function that creates a recorded job named `:context` containing
|
|
any value that needs to be shared across multiple workflow jobs.
|
|
|
|
## Examples
|
|
|
|
Add a context value to a workflow:
|
|
|
|
Workflow.new()
|
|
|> Workflow.put_context(%{user_id: 123, action: "process"})
|
|
|> Workflow.add(:job_a, WorkerA.new(%{}))
|
|
|> Workflow.add(:job_b, WorkerB.new(%{}))
|
|
|> Oban.insert_all()
|
|
|
|
Later, retrieve the context in a worker:
|
|
|
|
def process(job) do
|
|
context = Workflow.get_context(job)
|
|
# Use context map...
|
|
end
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec put_context(t(), map()) :: t()
|
|
def put_context(%__MODULE__{} = workflow, value) when is_map(value) do
|
|
opts = [
|
|
meta: %{context: true, recorded: true, return: Utils.encode64(value)},
|
|
worker: Oban.Pro.Workers.Context
|
|
]
|
|
|
|
changeset =
|
|
%{}
|
|
|> Job.new(opts)
|
|
|> Changeset.put_change(:state, "completed")
|
|
|> Changeset.put_change(:completed_at, DateTime.utc_now())
|
|
|
|
add(workflow, :context, changeset)
|
|
end
|
|
|
|
@doc """
|
|
Gets a workflow's context value.
|
|
|
|
This is a convenience function that internally calls `get_recorded/2` with the `:context` name.
|
|
|
|
## Examples
|
|
|
|
Retrieve the context within a worker:
|
|
|
|
def process(job) do
|
|
context = Workflow.get_context(job)
|
|
end
|
|
|
|
Retrieve the context with a workflow id:
|
|
|
|
Workflow.get_context("some-uuid-1234-5678")
|
|
|
|
Retrieve the context using a custom Oban instance name:
|
|
|
|
Workflow.get_context(MyApp.Oban, "some-uuid-1234-5678")
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec get_context(Oban.name(), job_or_wid()) :: nil | map()
|
|
def get_context(oban_name \\ Oban, job_or_wid) do
|
|
get_recorded(oban_name, job_or_wid, :context)
|
|
end
|
|
|
|
@doc """
|
|
Instantiate a new workflow from one or more existing workflow jobs.
|
|
|
|
## Examples
|
|
|
|
Append to a workflow seeded with all other jobs in the workflow:
|
|
|
|
jobs
|
|
|> Workflow.append()
|
|
|> Workflow.add(:d, WorkerD.new(%{}), deps: [:a])
|
|
|> Workflow.add(:e, WorkerE.new(%{}), deps: [:b])
|
|
|> Oban.insert_all()
|
|
|
|
Append to a workflow from a single job and bypass checking deps:
|
|
|
|
job
|
|
|> Workflow.append(check_deps: false)
|
|
|> Workflow.add(:d, WorkerD.new(%{}), deps: [:a])
|
|
|> Workflow.add(:e, WorkerE.new(%{}), deps: [:b])
|
|
|> Oban.insert_all()
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec append(Job.t() | [Job.t()], append_opts()) :: t()
|
|
def append(jobs, opts \\ [])
|
|
|
|
def append([%Job{meta: %{"workflow_id" => _} = meta} | _] = jobs, opts) do
|
|
{check, opts} = Keyword.pop(opts, :check_deps, true)
|
|
|
|
workflow_opts =
|
|
meta
|
|
|> Map.take(~w(workflow_id workflow_name))
|
|
|> Keyword.new(fn {key, val} -> {String.to_existing_atom(key), val} end)
|
|
|
|
# Preserve ancestor chain for nested sub-workflows
|
|
ancestors =
|
|
case meta do
|
|
%{"ancestor_ids" => ids} -> ids
|
|
%{"sup_workflow_id" => sup} -> [sup]
|
|
_ -> []
|
|
end
|
|
|
|
workflow =
|
|
opts
|
|
|> Keyword.merge(workflow_opts)
|
|
|> new()
|
|
|
|
opts = Map.put(workflow.opts, :ancestor_ids, ancestors)
|
|
|
|
%{workflow | check_deps: check, names: MapSet.new(jobs, & &1.meta["name"]), opts: opts}
|
|
end
|
|
|
|
def append(job, opts), do: append([job], opts)
|
|
|
|
@doc """
|
|
Attaches a sub-workflow to a previously defined grafting point.
|
|
|
|
Grafting points are defined using `add_graft/4` and serve as placeholders where sub-workflows can be
|
|
attached dynamically during execution. This function must be called from within a worker's
|
|
`process/1` callback, and only from a job that was defined using `add_graft/4`.
|
|
|
|
## Examples
|
|
|
|
Graft a sub-workflow using a cascade:
|
|
|
|
def process(job) do
|
|
user_ids = active_user_ids(job.args["account_id"])
|
|
|
|
{user_ids, &build_user_workflow/2}
|
|
|> Workflow.apply_graft()
|
|
|> Oban.insert_all()
|
|
|
|
:ok
|
|
end
|
|
|
|
Graft a sub-workflow using a workflow:
|
|
|
|
def process(job) do
|
|
workflow =
|
|
job.args["account_id"]
|
|
|> active_user_ids()
|
|
|> Enum.reduce(Workflow.new(), fn user_id, workflow ->
|
|
Workflow.add(workflow, user_id, UserWorker.new(%{user_id: user_id}))
|
|
end)
|
|
|
|
workflow
|
|
|> Workflow.apply_graft()
|
|
|> Oban.insert_all()
|
|
|
|
:ok
|
|
end
|
|
|
|
Add workflow-specific options when grafting:
|
|
|
|
{user_ids, &build_user_workflow/2}
|
|
|> Workflow.apply_graft(workflow_name: "my-workflow")
|
|
|> Oban.insert_all()
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec apply_graft(t() | cascade_capture(), new_opts() | add_cascade_opts()) :: t()
|
|
def apply_graft(cascade_or_workflow, opts \\ [])
|
|
|
|
def apply_graft({_enum, fun} = cascade, opts) when is_function(fun) do
|
|
{job, graft_id, graft_name} = extract_graft!()
|
|
|
|
job
|
|
|> append(check_deps: false)
|
|
|> add_cascade(graft_name, cascade, Keyword.put(opts, :workflow_id, graft_id))
|
|
end
|
|
|
|
def apply_graft(%__MODULE__{} = workflow, opts) do
|
|
{job, graft_id, graft_name} = extract_graft!()
|
|
|
|
root_id =
|
|
case job.meta["workflow_id"] do
|
|
"grafted_" <> _ = parent_graft_id -> parent_graft_id
|
|
_ -> graft_id
|
|
end
|
|
|
|
changesets =
|
|
Enum.map(workflow.changesets, fn changeset ->
|
|
if get_in(changeset.changes, [:meta, :sup_workflow_id]) do
|
|
changeset
|
|
else
|
|
Changeset.update_change(changeset, :meta, &Map.put(&1, :workflow_id, root_id))
|
|
end
|
|
end)
|
|
|
|
workflow = %{workflow | id: root_id, changesets: changesets}
|
|
|
|
job
|
|
|> append(check_deps: false)
|
|
|> add_workflow(graft_name, workflow, opts)
|
|
end
|
|
|
|
defp extract_graft! do
|
|
case Process.get(:oban_processing) do
|
|
{_module, %{meta: %{"graft_id" => _}} = job, _opts} ->
|
|
%{"graft_id" => graft_id, "graft_name" => graft_name} = job.meta
|
|
|
|
{job, graft_id, graft_name}
|
|
|
|
{_module, _job, _opts} ->
|
|
raise "current job isnt't marked for grafting with add_graft/4"
|
|
|
|
_ ->
|
|
raise "current job isn't available, graft/1 must be called from process/1"
|
|
end
|
|
end
|
|
|
|
@doc false
|
|
def all_jobs(job_or_wid), do: all_jobs(Oban, job_or_wid, [])
|
|
|
|
@doc false
|
|
def all_jobs(job_or_wid, opts) when is_job_or_wid(job_or_wid) and is_list(opts) do
|
|
all_jobs(Oban, job_or_wid, opts)
|
|
end
|
|
|
|
def all_jobs(oban_name, job_or_wid) when is_job_or_wid(job_or_wid) do
|
|
all_jobs(oban_name, job_or_wid, [])
|
|
end
|
|
|
|
@doc """
|
|
Get all jobs for a workflow, optionally filtered by upstream deps.
|
|
|
|
## Examples
|
|
|
|
Retrieve all workflow jobs within a `process/1` function:
|
|
|
|
def process(job) do
|
|
job
|
|
|> Workflow.all_jobs()
|
|
|> do_things_with_jobs()
|
|
|
|
:ok
|
|
end
|
|
|
|
Retrieve all of the current job's deps:
|
|
|
|
jobs = Workflow.all_jobs(job, only_deps: true)
|
|
|
|
Retrieve an explicit list of dependencies:
|
|
|
|
[job_a, job_b] = Workflow.all_jobs(job, names: [:a, :b])
|
|
|
|
Include sub workflow jobs:
|
|
|
|
[sub_a, sub_b] = Workflow.all_jobs(job, names: [:sub], with_subs: true)
|
|
|
|
Retrieve all jobs using a `workflow_id`:
|
|
|
|
jobs = Workflow.all_jobs("some-uuid-1234-5678")
|
|
|
|
Retrieve only some named jobs using a `workflow_id`:
|
|
|
|
[job_a, job_b] = Workflow.all_jobs("some-uuid-1234-5678", deps: [:a, :b])
|
|
|
|
Retrieve all jobs using a `workflow_id` and a custom Oban instance name:
|
|
|
|
jobs = Workflow.all_jobs(MyApp.Oban, "some-uuid-1234-5678")
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec all_jobs(Oban.name(), job_or_wid(), fetch_opts()) :: [Job.t()]
|
|
def all_jobs(oban_name, job_or_wid, opts)
|
|
|
|
def all_jobs(_oban_name, %Job{conf: conf, meta: meta}, opts) do
|
|
%{"deps" => deps, "workflow_id" => workflow_id} = meta
|
|
|
|
{query_opts, opts} = Keyword.split(opts, ~w(names only_deps with_subs)a)
|
|
|
|
Repo.all(conf, workflow_query(workflow_id, deps, query_opts), opts)
|
|
end
|
|
|
|
def all_jobs(oban_name, workflow_id, opts) when is_binary(workflow_id) do
|
|
if Keyword.has_key?(opts, :only_deps) do
|
|
raise RuntimeError, ":only_deps isn't a valid option with a workflow_id"
|
|
end
|
|
|
|
{query_opts, opts} = Keyword.split(opts, ~w(names with_subs)a)
|
|
|
|
oban_name
|
|
|> Oban.config()
|
|
|> Repo.all(workflow_query(workflow_id, [], query_opts), opts)
|
|
end
|
|
|
|
@doc false
|
|
def all_recorded(job_or_wid), do: all_recorded(Oban, job_or_wid, [])
|
|
|
|
@doc false
|
|
def all_recorded(job_or_wid, opts) when is_job_or_wid(job_or_wid) and is_list(opts) do
|
|
all_recorded(Oban, job_or_wid, opts)
|
|
end
|
|
|
|
def all_recorded(oban_name, job_or_wid) when is_job_or_wid(job_or_wid) do
|
|
all_recorded(oban_name, job_or_wid, [])
|
|
end
|
|
|
|
@doc """
|
|
Get all recordings for workflow jobs, optionally filtered by name or relationship.
|
|
|
|
The result is a map of the job's name and recorded output. Unrecorded jobs, or those without a
|
|
recording will be present in the map with a a `nil` value.
|
|
|
|
## Examples
|
|
|
|
Retrieve results for all workflow jobs within a `process/1` function:
|
|
|
|
def process(job) do
|
|
%{"a" => job_a_return, "b" => job_b_return} = Workflow.all_recorded(job)
|
|
|
|
:ok
|
|
end
|
|
|
|
Retrieve recordings for all of the current job's deps:
|
|
|
|
recorded = Workflow.all_recorded(job, only_deps: true)
|
|
|
|
Retrieve recordings for a list of dependencies:
|
|
|
|
%{"a" => _, "b" => _} = Workflow.all_recorded(job, names: [:a, :b])
|
|
|
|
Retrieve recordings for some named jobs using the `workflow_id`:
|
|
|
|
%{"a" => _, "b" => _} = Workflow.all_recorded("some-uuid-1234-5678", names: [:a, :b])
|
|
|
|
Retrieve recordings including results from grafted sub-workflows:
|
|
|
|
recorded = Workflow.all_recorded(job, with_subs: true)
|
|
|
|
Retrieve recordings for a specific grafted sub-workflow by name:
|
|
|
|
%{"users" => user_results} = Workflow.all_recorded(job, names: [:users], with_subs: true)
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec all_recorded(Oban.name(), job_or_wid(), fetch_opts()) :: map()
|
|
def all_recorded(oban_name, job_or_wid, opts)
|
|
|
|
def all_recorded(_oban_name, %Job{conf: conf, meta: meta}, opts) do
|
|
%{"deps" => deps, "workflow_id" => workflow_id} = meta
|
|
|
|
all_recorded(conf, workflow_id, deps, opts)
|
|
end
|
|
|
|
def all_recorded(%Config{} = conf, workflow_id, opts) when is_binary(workflow_id) do
|
|
if Keyword.has_key?(opts, :only_deps) do
|
|
raise RuntimeError, ":only_deps isn't a valid option with a workflow_id"
|
|
end
|
|
|
|
all_recorded(conf, workflow_id, [], opts)
|
|
end
|
|
|
|
def all_recorded(oban_name, workflow_id, opts) when is_binary(workflow_id) do
|
|
oban_name
|
|
|> Oban.config()
|
|
|> all_recorded(workflow_id, [], opts)
|
|
end
|
|
|
|
defp all_recorded(conf, workflow_id, deps, opts) when is_binary(workflow_id) do
|
|
{query_opts, opts} = Keyword.split(opts, ~w(names only_deps with_subs)a)
|
|
|
|
with_subs = Keyword.get(query_opts, :with_subs, false)
|
|
|
|
query =
|
|
workflow_id
|
|
|> workflow_query(deps, query_opts)
|
|
|> select([j], j.meta)
|
|
|
|
conf
|
|
|> Repo.all(query, opts)
|
|
|> Enum.reject(&Map.get(&1, "context"))
|
|
|> Enum.reduce(%{}, fn meta, acc ->
|
|
case meta do
|
|
%{"name" => name, "sub_name" => work_name} when with_subs ->
|
|
acc
|
|
|> Map.put_new(work_name, %{})
|
|
|> put_in([work_name, name], extract_recorded(meta))
|
|
|
|
%{"name" => name} ->
|
|
Map.put(acc, name, extract_recorded(meta))
|
|
end
|
|
end)
|
|
end
|
|
|
|
@doc false
|
|
def cancel_jobs(job_or_wid), do: cancel_jobs(Oban, job_or_wid, [])
|
|
|
|
@doc false
|
|
def cancel_jobs(job_or_wid, opts) when is_job_or_wid(job_or_wid) and is_list(opts) do
|
|
cancel_jobs(Oban, job_or_wid, opts)
|
|
end
|
|
|
|
def cancel_jobs(oban_name, job_or_wid) when is_job_or_wid(job_or_wid) do
|
|
cancel_jobs(oban_name, job_or_wid, [])
|
|
end
|
|
|
|
@doc """
|
|
Cancel one or more workflow jobs.
|
|
|
|
This uses `Oban.cancel_all_jobs/2` internally, and adheres to the same cancellation rules.
|
|
Namely, it won't touch jobs in a `completed`, `cancelled`, or `discarded` state.
|
|
|
|
Cancelling jobs via this function bulk-updates their state in the database without loading them
|
|
into memory or resolving their workers. As a result, **worker `after_process/3` hooks are not
|
|
called** for cancelled jobs. If you need to execute cleanup logic or other side effects when jobs
|
|
are cancelled, consider using workflow callbacks like `after_cancelled/2` or implement cleanup
|
|
logic at the application level.
|
|
|
|
## Examples
|
|
|
|
Cancel jobs with a workflow `job` from a `process/1` function:
|
|
|
|
def process(job) do
|
|
if should_stop_processing?(job.args) do
|
|
Workflow.cancel_jobs(job)
|
|
else
|
|
...
|
|
end
|
|
end
|
|
|
|
Cancel specific workflow jobs:
|
|
|
|
Workflow.cancel_jobs("some-uuid-1234-5678", names: [:a, :b])
|
|
|
|
Cancel all jobs in a workflow by `workflow_id`:
|
|
|
|
Workflow.cancel_jobs("some-uuid-1234-5678")
|
|
|
|
Cancel jobs from anywhere with a custom Oban instance name:
|
|
|
|
Workflow.cancel_jobs(MyApp.Oban, "some-uuid-1234-5678")
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec cancel_jobs(Oban.name(), job_or_wid(), cancel_opts()) :: {:ok, integer()}
|
|
def cancel_jobs(oban_name, job_or_wid, opts)
|
|
|
|
def cancel_jobs(_oban_name, %Job{conf: conf, meta: %{"workflow_id" => wid}}, opts) do
|
|
cancel_jobs(conf.name, wid, opts)
|
|
end
|
|
|
|
def cancel_jobs(oban_name, wid, opts) when is_binary(wid) do
|
|
query_opts =
|
|
opts
|
|
|> Keyword.take([:names, :with_subs])
|
|
|> Keyword.put_new(:with_subs, true)
|
|
|
|
Oban.cancel_all_jobs(oban_name, workflow_query(wid, [], query_opts))
|
|
end
|
|
|
|
@doc """
|
|
Retry all jobs in a workflow.
|
|
|
|
By default, this function will retry all jobs that aren't `executing` or `available`. Jobs with
|
|
dependencies will be placed on hold to run again once their dependencies are complete.
|
|
|
|
## Examples
|
|
|
|
Retry only failed jobs in a workflow from within a process function:
|
|
|
|
def process(job) do
|
|
Workflow.retry_jobs(job)
|
|
end
|
|
|
|
Retry a workflow by ID:
|
|
|
|
Workflow.retry_jobs("some-uuid-1234-5678")
|
|
|
|
Retry a workflow with a custom Oban instance:
|
|
|
|
Workflow.retry_jobs(MyApp.Oban, "some-uuid-1234-5678")
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec retry_jobs(Oban.name(), job_or_wid(), keyword()) :: {:ok, integer()}
|
|
def retry_jobs(oban_name \\ Oban, job_or_wid, opts \\ [])
|
|
|
|
def retry_jobs(_oban_name, %Job{conf: conf, meta: %{"workflow_id" => wid}}, opts) do
|
|
retry_jobs(conf.name, wid, opts)
|
|
end
|
|
|
|
def retry_jobs(oban_name, wid, opts) when is_binary(wid) do
|
|
conf = Oban.config(oban_name)
|
|
|
|
Repo.transaction(conf, fn ->
|
|
jobs = all_jobs(oban_name, wid, opts)
|
|
|
|
{with_deps, wout_deps} =
|
|
jobs
|
|
|> Enum.reject(& &1.meta["context"])
|
|
|> Enum.split_with(fn %{meta: meta} -> Enum.any?(meta["deps"]) end)
|
|
|
|
base = where(Job, [j], j.state not in ~w(available executing))
|
|
|
|
if Enum.any?(wout_deps) do
|
|
query =
|
|
base
|
|
|> where([j], j.id in ^Enum.map(wout_deps, & &1.id))
|
|
|> update([j],
|
|
set: [
|
|
state: "available",
|
|
max_attempts: fragment("GREATEST(?, ? + 1)", j.max_attempts, j.attempt),
|
|
scheduled_at: ^DateTime.utc_now(),
|
|
completed_at: nil,
|
|
cancelled_at: nil,
|
|
discarded_at: nil
|
|
]
|
|
)
|
|
|
|
Repo.update_all(conf, query, [])
|
|
end
|
|
|
|
if Enum.any?(with_deps) do
|
|
query =
|
|
base
|
|
|> where([j], j.id in ^Enum.map(with_deps, & &1.id))
|
|
|> update([j],
|
|
set: [
|
|
state: "suspended",
|
|
max_attempts: fragment("GREATEST(?, ? + 1)", j.max_attempts, j.attempt),
|
|
completed_at: nil,
|
|
cancelled_at: nil,
|
|
discarded_at: nil
|
|
]
|
|
)
|
|
|
|
Repo.update_all(conf, query, [])
|
|
end
|
|
|
|
length(jobs)
|
|
end)
|
|
end
|
|
|
|
@doc """
|
|
Deliver a signal to one or more named jobs in a workflow.
|
|
|
|
A thin wrapper over `Oban.Pro.Worker.signal/3` that resolves workflow job names before
|
|
delegating. See the `Oban.Pro.Worker` docs for the underlying semantics.
|
|
|
|
## Examples
|
|
|
|
Signal a single named job:
|
|
|
|
Workflow.signal(workflow_id, :approval, %{decision: "approved"})
|
|
|
|
Broadcast a signal to several named jobs:
|
|
|
|
Workflow.signal(workflow_id, [:reviewer_a, :reviewer_b], %{decision: "approved"})
|
|
|
|
Target a non-default Oban instance:
|
|
|
|
Workflow.signal(MyApp.Oban, workflow_id, :approval, %{decision: "approved"})
|
|
"""
|
|
@doc since: "1.7.0"
|
|
@spec signal(binary(), name() | [name()], term()) :: :ok
|
|
def signal(workflow_id, name_or_names, payload) when is_binary(workflow_id) do
|
|
signal(Oban, workflow_id, name_or_names, payload)
|
|
end
|
|
|
|
@spec signal(Oban.name(), binary(), name() | [name()], term()) :: :ok
|
|
def signal(oban_name, workflow_id, name_or_names, payload) when is_binary(workflow_id) do
|
|
conf = Oban.config(oban_name)
|
|
|
|
names =
|
|
name_or_names
|
|
|> List.wrap()
|
|
|> Enum.map(&to_string/1)
|
|
|
|
query =
|
|
Job
|
|
|> where([j], in_workflow(j.meta, ^workflow_id))
|
|
|> where([j], fragment("?->>'name' = ANY(?)", j.meta, ^names))
|
|
|> select([j], j.id)
|
|
|
|
Oban.Pro.Worker.signal(oban_name, Repo.all(conf, query), payload)
|
|
end
|
|
|
|
@doc """
|
|
Get a single workflow job by name.
|
|
|
|
## Examples
|
|
|
|
Get the workflow job named `:step_1` from within `process/`:
|
|
|
|
def process(job) do
|
|
case Workflow.get_job(job, :step_1) do
|
|
nil -> ...
|
|
dep_job -> use_the_job(dep_job)
|
|
end
|
|
end
|
|
|
|
Get a named workflow job with the `workflow_id`:
|
|
|
|
Workflow.get_job("some-uuid-1234-5678", :step_2)
|
|
|
|
Get a named workflow job with the `workflow_id` and a custom Oban instance name:
|
|
|
|
Workflow.get_job(MyApp.Oban, "some-uuid-1234-5678", :step_2)
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec get_job(Oban.name(), job_or_wid(), name()) :: nil | Job.t()
|
|
def get_job(oban_name \\ Oban, job_or_wid, deps_name)
|
|
|
|
def get_job(_oban_name, %Job{conf: conf, meta: %{"workflow_id" => wid}}, deps_name) do
|
|
get_job(conf.name, wid, deps_name)
|
|
end
|
|
|
|
def get_job(oban_name, wid, deps_name) when is_binary(wid) do
|
|
case all_jobs(oban_name, wid, names: [deps_name]) do
|
|
[job | _] -> job
|
|
[] -> nil
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Fetch the recorded output from a workflow job.
|
|
|
|
This function provides an optimized way to get recorded output with a single query.
|
|
|
|
A `nil` value is returned when a dependency is incomplete, missing, unrecorded, or lacks
|
|
recorded output.
|
|
|
|
## Examples
|
|
|
|
Get the recorded output from `:step_1` within `process/`:
|
|
|
|
def process(job) do
|
|
case Workflow.get_recorded(job, :step_1) do
|
|
nil -> ...
|
|
val -> use_recorded_output(val)
|
|
end
|
|
end
|
|
|
|
Get recorded output using the workflow id:
|
|
|
|
Workflow.get_recorded("some-uuid-1234-5678", :step_2)
|
|
|
|
Get recorded output using the `workflow_id` and a custom Oban instance name:
|
|
|
|
Workflow.get_recorded(MyApp.Oban, "some-uuid-1234-5678", :step_2)
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec get_recorded(Oban.name(), job_or_wid(), name()) :: any()
|
|
def get_recorded(oban_name \\ Oban, job_or_wid, deps_name)
|
|
|
|
def get_recorded(_oban_name, %Job{conf: conf, meta: %{"workflow_id" => wid}}, deps_name) do
|
|
get_recorded(conf, wid, deps_name)
|
|
end
|
|
|
|
def get_recorded(%Config{} = conf, wid, deps_name) when is_binary(wid) do
|
|
query =
|
|
wid
|
|
|> workflow_query([], names: [deps_name])
|
|
|> select([j], j.meta)
|
|
|
|
conf
|
|
|> Repo.one(query)
|
|
|> extract_recorded()
|
|
end
|
|
|
|
def get_recorded(oban_name, wid, deps_name) when is_binary(wid) do
|
|
oban_name
|
|
|> Oban.config()
|
|
|> get_recorded(wid, deps_name)
|
|
end
|
|
|
|
@doc false
|
|
def stream_jobs(job_or_wid), do: stream_jobs(Oban, job_or_wid, [])
|
|
|
|
@doc false
|
|
def stream_jobs(job_or_wid, opts) when is_job_or_wid(job_or_wid) and is_list(opts) do
|
|
stream_jobs(Oban, job_or_wid, opts)
|
|
end
|
|
|
|
def stream_jobs(oban_name, job_or_wid) when is_job_or_wid(job_or_wid) do
|
|
stream_jobs(oban_name, job_or_wid, [])
|
|
end
|
|
|
|
@doc """
|
|
Stream all jobs for a workflow.
|
|
|
|
This function behaves identically to `all_jobs/3`, except it streams jobs lazily from within a
|
|
`Repo` transaction.
|
|
|
|
## Examples
|
|
|
|
Stream with filtering to only preserve `completed` jobs:
|
|
|
|
def process(job) do
|
|
{:ok, workflow_jobs} =
|
|
MyApp.Repo.transaction(fn ->
|
|
job
|
|
|> Workflow.stream_jobs()
|
|
|> Stream.filter(& &1.state == "completed")
|
|
|> Enum.to_list()
|
|
end)
|
|
|
|
do_things_with_jobs(workflow_jobs)
|
|
|
|
:ok
|
|
end
|
|
|
|
Stream workflow jobs from anywhere using the `workflow_id`:
|
|
|
|
MyApp.Repo.transaction(fn ->
|
|
"some-uuid-1234-5678"
|
|
|> Workflow.stream_jobs()
|
|
|> Enum.map(& &1.args["account_id"])
|
|
end)
|
|
|
|
Stream workflow jobs using a custom Oban instance name:
|
|
|
|
MyApp.Repo.transaction(fn ->
|
|
MyApp.Oban
|
|
|> Workflow.stream_jobs("some-uuid-1234-5678")
|
|
|> Enum.map(& &1.args["account_id"])
|
|
end)
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec stream_jobs(Oban.name(), Job.t() | String.t(), fetch_opts()) :: Enum.t()
|
|
def stream_jobs(oban_name, job_or_wid, opts)
|
|
|
|
def stream_jobs(_oban_name, %Job{conf: conf, meta: meta}, opts) do
|
|
%{"deps" => deps, "workflow_id" => wid} = meta
|
|
|
|
{query_opts, opts} = Keyword.split(opts, [:names, :only_deps])
|
|
|
|
Repo.stream(conf, workflow_query(wid, deps, query_opts), opts)
|
|
end
|
|
|
|
def stream_jobs(oban_name, wid, opts) when is_binary(wid) do
|
|
if Keyword.has_key?(opts, :only_deps) do
|
|
raise RuntimeError, ":only_deps isn't a valid option with a workflow_id"
|
|
end
|
|
|
|
{query_opts, opts} = Keyword.split(opts, [:names])
|
|
|
|
oban_name
|
|
|> Oban.config()
|
|
|> Repo.stream(workflow_query(wid, [], query_opts), opts)
|
|
end
|
|
|
|
@doc """
|
|
Get detailed status information about a workflow.
|
|
|
|
Returns a map with the following fields:
|
|
|
|
- `:id` - The workflow's unique id
|
|
- `:name` - The name of the workflow, if it has one
|
|
- `:total` - Total number of jobs in the workflow
|
|
- `:state` - Overall status of the workflow (e.g. executing, completed, cancelled, discarded)
|
|
- `:counts` - Map of job state counts (e.g. `%{completed: 10, available: 2}`)
|
|
- `:duration` - Elapsed time of the workflow in milliseconds (nil if not yet ran)
|
|
- `:started_at` - When the first job in the workflow was inserted
|
|
- `:stopped_at` - When the last job in the workflow finished
|
|
- `:subs` - A map of status output for sub-worfklows where the key is the sub name and the value
|
|
is a `status/1` output
|
|
|
|
An example of the full output:
|
|
|
|
```elixir
|
|
%{
|
|
id: "some-uuid-1234-5678",
|
|
name: "my-workflow",
|
|
total: 3,
|
|
state: :completed,
|
|
duration: 72,
|
|
counts: %{
|
|
cancelled: 0,
|
|
available: 0,
|
|
completed: 3,
|
|
scheduled: 0,
|
|
discarded: 0,
|
|
executing: 0,
|
|
retryable: 0
|
|
},
|
|
started_at: ~U[2025-04-10 16:34:34.114730Z],
|
|
stopped_at: ~U[2025-04-10 16:34:34.186935Z],
|
|
subs: %{}
|
|
}
|
|
```
|
|
|
|
## Examples
|
|
|
|
Get a workflow's status using a job:
|
|
|
|
status = Workflow.status(job)
|
|
|
|
Get a workflow's status using the workflow_id:
|
|
|
|
status = Workflow.status("some-uuid-1234-5678")
|
|
|
|
Get a workflow's status using a custom Oban instance name:
|
|
|
|
status = Workflow.status(MyApp.Oban, "some-uuid-1234-5678")
|
|
"""
|
|
@doc since: "1.6.0"
|
|
@spec status(Oban.name(), job_or_wid()) :: status()
|
|
def status(oban_name \\ Oban, job_or_wid)
|
|
|
|
def status(_oban_name, %Job{conf: conf, meta: %{"workflow_id" => wid}}) do
|
|
status(conf, wid)
|
|
end
|
|
|
|
def status(%Config{} = conf, workflow_id) when is_binary(workflow_id) do
|
|
query =
|
|
Job
|
|
|> where([j], in_workflow(j.meta, ^workflow_id))
|
|
|> group_by([j], [j.meta["workflow_id"], j.meta["workflow_name"]])
|
|
|> select([j], %{
|
|
id: j.meta["workflow_id"],
|
|
name: j.meta["workflow_name"],
|
|
ignore_cancelled: fragment("array_agg(?)", j.meta["ignore_cancelled"]),
|
|
ignore_discarded: fragment("array_agg(?)", j.meta["ignore_discarded"]),
|
|
states: fragment("array_agg(?)", j.state),
|
|
min_scheduled: min(j.inserted_at),
|
|
max_completed: max(j.completed_at),
|
|
max_cancelled: max(j.cancelled_at),
|
|
max_discarded: max(j.discarded_at)
|
|
})
|
|
|
|
case Repo.one(conf, query) do
|
|
nil ->
|
|
%{
|
|
id: workflow_id,
|
|
name: nil,
|
|
total: 0,
|
|
state: :unknown,
|
|
counts: Map.new(Job.states(), &{&1, 0}),
|
|
duration: nil,
|
|
started_at: nil,
|
|
stopped_at: nil,
|
|
subs: %{}
|
|
}
|
|
|
|
result ->
|
|
expand_status(conf, workflow_id, result)
|
|
end
|
|
end
|
|
|
|
def status(oban_name, workflow_id) when is_binary(workflow_id) do
|
|
oban_name
|
|
|> Oban.config()
|
|
|> status(workflow_id)
|
|
end
|
|
|
|
defp expand_status(conf, workflow_id, result) do
|
|
total = length(result.states)
|
|
empty = Map.new(Job.states(), &{&1, 0})
|
|
|
|
counts =
|
|
[result.states, result.ignore_cancelled, result.ignore_discarded]
|
|
|> Enum.zip()
|
|
|> Enum.reduce(empty, fn tuple, acc ->
|
|
state =
|
|
case tuple do
|
|
{"cancelled", true, _} -> :completed
|
|
{"discarded", _, true} -> :completed
|
|
{state, _ican, _idisc} -> String.to_existing_atom(state)
|
|
end
|
|
|
|
Map.update(acc, state, 1, &(&1 + 1))
|
|
end)
|
|
|
|
state =
|
|
cond do
|
|
counts.completed + counts.cancelled + counts.discarded < total -> :executing
|
|
counts.cancelled > 0 -> :cancelled
|
|
counts.discarded > 0 -> :discarded
|
|
true -> :completed
|
|
end
|
|
|
|
stopped_at =
|
|
[result.max_completed, result.max_cancelled, result.max_discarded]
|
|
|> Enum.reject(&is_nil/1)
|
|
|> Enum.max(fn -> nil end)
|
|
|
|
duration =
|
|
if stopped_at do
|
|
DateTime.diff(stopped_at, result.min_scheduled, :millisecond)
|
|
else
|
|
nil
|
|
end
|
|
|
|
subs =
|
|
Job
|
|
|> where([j], fragment("? \\? 'sup_workflow_id'", j.meta))
|
|
|> where([j], fragment("?->>'sup_workflow_id' = ?", j.meta, ^workflow_id))
|
|
|> where([j], fragment("?->>'workflow_id' <> ?", j.meta, ^workflow_id))
|
|
|> select([j], %{wid: j.meta["workflow_id"], name: j.meta["sub_name"]})
|
|
|> distinct(true)
|
|
|> then(&Repo.all(conf, &1))
|
|
|> Map.new(fn %{wid: wid, name: name} -> {name, status(conf, wid)} end)
|
|
|
|
%{
|
|
id: result.id,
|
|
name: result.name,
|
|
total: total,
|
|
state: state,
|
|
counts: counts,
|
|
duration: duration,
|
|
started_at: result.min_scheduled,
|
|
stopped_at: stopped_at,
|
|
subs: subs
|
|
}
|
|
end
|
|
|
|
@doc """
|
|
Converts the given workflow to a `:digraph`.
|
|
|
|
The resulting digraph can be explored, evaluated, and then be converted to a number of other
|
|
formats.
|
|
|
|
## Examples
|
|
|
|
Generate a digraph from a workflow:
|
|
|
|
graph = Workflow.to_graph(workflow)
|
|
|
|
Check the path between jobs in a workflow:
|
|
|
|
workflow
|
|
|> Workflow.to_graph()
|
|
|> :digraph.get_path("step_1", "step_5")
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec to_graph(flow :: t()) :: :digraph.graph()
|
|
def to_graph(%__MODULE__{changesets: changesets, subs: subs}) do
|
|
graph = :digraph.new([:acyclic])
|
|
|
|
changesets
|
|
|> Enum.group_by(& &1.changes.meta[:sub_name])
|
|
|> Enum.each(fn
|
|
{nil, changesets} ->
|
|
add_graph_nodes(graph, changesets)
|
|
|
|
{sub, changesets} ->
|
|
sub_graph =
|
|
[:acyclic]
|
|
|> :digraph.new()
|
|
|> add_graph_nodes(changesets)
|
|
|
|
:digraph.add_vertex(graph, sub, sub_graph)
|
|
end)
|
|
|
|
for %{changes: %{meta: meta}} <- changesets, dep <- meta.deps do
|
|
# This would also create edges in the top graph for values that shouldn't be there.
|
|
case dep do
|
|
[wid, "*"] ->
|
|
sub_name = Enum.find_value(subs, &if(elem(&1, 1) == wid, do: elem(&1, 0)))
|
|
dep_name = meta[:sub_name] || meta[:name]
|
|
|
|
:digraph.add_edge(graph, sub_name, dep_name)
|
|
|
|
[_id, dep] ->
|
|
:digraph.add_edge(graph, dep, meta.sub_name)
|
|
|
|
dep ->
|
|
:digraph.add_edge(graph, dep, meta.name)
|
|
end
|
|
end
|
|
|
|
graph
|
|
end
|
|
|
|
defp add_graph_nodes(graph, changesets) do
|
|
for %{changes: %{meta: meta, worker: worker}} <- changesets do
|
|
label = Map.get(meta, :decorated_name, worker)
|
|
|
|
:digraph.add_vertex(graph, meta.name, label)
|
|
end
|
|
|
|
graph
|
|
end
|
|
|
|
@doc """
|
|
Converts the given workflow to a [graphviz](https://graphviz.org/) `dot` digraph.
|
|
|
|
## Examples
|
|
|
|
Generate a dot digraph:
|
|
|
|
Workflow.to_dot(workflow)
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec to_dot(flow :: t()) :: String.t()
|
|
def to_dot(%__MODULE__{} = workflow) do
|
|
graph = to_graph(workflow)
|
|
nodes = :digraph.vertices(graph)
|
|
edges = :digraph.edges(graph)
|
|
|
|
nodes_dot = visit_nodes(nodes, graph, :dot)
|
|
edges_dot = visit_edges(edges, graph, :dot)
|
|
|
|
"""
|
|
digraph {
|
|
node [shape=box];
|
|
#{nodes_dot}
|
|
#{edges_dot}
|
|
}
|
|
"""
|
|
end
|
|
|
|
@doc """
|
|
Generate a [Mermaid](https://mermaid.js.org/) flowchart in top-down orientation from a workflow.
|
|
|
|
## Examples
|
|
|
|
Generate a flowchart:
|
|
|
|
Workflow.to_mermaid(workflow)
|
|
"""
|
|
@doc since: "1.5.0"
|
|
@spec to_mermaid(flow :: t()) :: String.t()
|
|
def to_mermaid(%__MODULE__{} = workflow) do
|
|
graph = to_graph(workflow)
|
|
nodes = :digraph.vertices(graph)
|
|
edges = :digraph.edges(graph)
|
|
|
|
nodes_mermaid = visit_nodes(nodes, graph, :mermaid)
|
|
edges_mermaid = visit_edges(edges, graph, :mermaid)
|
|
|
|
"""
|
|
flowchart TD
|
|
#{nodes_mermaid}
|
|
#{edges_mermaid}
|
|
"""
|
|
end
|
|
|
|
@doc false
|
|
def rescue_workflows(conf, opts \\ []) do
|
|
{chunk_size, opts} = Keyword.pop(opts, :chunk_size, 100)
|
|
{since_secs, opts} = Keyword.pop(opts, :max_age, 60)
|
|
|
|
threshold = DateTime.add(DateTime.utc_now(), -since_secs, :second)
|
|
|
|
query =
|
|
Schema
|
|
|> select([w], w.id)
|
|
|> where([w], w.suspended > 0)
|
|
|> where([w], coalesce(w.started_at, w.inserted_at) < ^threshold)
|
|
|
|
workflow_ids = Repo.all(conf, query, opts)
|
|
|
|
workflow_ids =
|
|
if Utils.has_legacy_workflows?(conf) do
|
|
legacy_query =
|
|
Job
|
|
|> select([j], fragment("?->>'workflow_id'", j.meta))
|
|
|> where([j], j.state == "scheduled")
|
|
|> where([j], fragment("? \\? 'workflow_id'", j.meta))
|
|
|> where([j], fragment("?->>'on_hold' = 'true'", j.meta))
|
|
|> distinct(true)
|
|
|
|
legacy_ids = Repo.all(conf, legacy_query, opts)
|
|
|
|
Enum.uniq(workflow_ids ++ legacy_ids)
|
|
else
|
|
workflow_ids
|
|
end
|
|
|
|
workflow_ids
|
|
|> Enum.chunk_every(chunk_size)
|
|
|> Enum.each(&on_flush(&1, conf))
|
|
end
|
|
|
|
# Construction Helpers
|
|
|
|
defp prevent_chain!(changeset, name) do
|
|
if Chain.chain?(changeset) do
|
|
raise "#{inspect(name)} is a chain, which isn't allowed in a workflow"
|
|
end
|
|
end
|
|
|
|
defp prevent_dupe!(%{names: names, subs: subs}, name) do
|
|
if MapSet.member?(names, name) or is_map_key(subs, name) do
|
|
raise "#{inspect(name)} is already a member of the workflow"
|
|
end
|
|
end
|
|
|
|
defp confirm_deps!(%{check_deps: true, names: names, grafts: grafts}, [_ | _] = deps) do
|
|
Enum.each(deps, fn
|
|
[_wd, "*"] ->
|
|
:ok
|
|
|
|
[_wd, name] ->
|
|
if not MapSet.member?(names, name) do
|
|
raise "deps #{inspect(name)} is not a member of the workflow"
|
|
end
|
|
|
|
name when is_binary(name) ->
|
|
if not (MapSet.member?(names, name) or is_map_key(grafts, name)) do
|
|
raise "deps #{inspect(name)} is not a member of the workflow"
|
|
end
|
|
end)
|
|
end
|
|
|
|
defp confirm_deps!(_workflow, _deps), do: :ok
|
|
|
|
# Visualization Helpers
|
|
|
|
defp visit_nodes(nodes, graph, mode, prefix \\ nil) do
|
|
Enum.map_join(nodes, "\n", fn node ->
|
|
{_node, label} = :digraph.vertex(graph, node)
|
|
|
|
if is_tuple(label) do
|
|
sub_nodes = visit_nodes(:digraph.vertices(label), label, mode, "#{node}_")
|
|
sub_edges = visit_edges(:digraph.edges(label), label, mode)
|
|
|
|
format_subg(mode, node, sub_nodes, sub_edges)
|
|
else
|
|
format_node(mode, "#{prefix}#{node}", label)
|
|
end
|
|
end)
|
|
end
|
|
|
|
defp visit_edges(edges, graph, mode) do
|
|
edges
|
|
|> Enum.map(fn edge ->
|
|
{_, in_vert, out_vert, _} = :digraph.edge(graph, edge)
|
|
|
|
{in_vert, out_vert}
|
|
end)
|
|
|> Enum.uniq()
|
|
|> Enum.map_join("\n ", &format_edge(mode, &1))
|
|
end
|
|
|
|
defp format_node(:dot, node, label) do
|
|
~s|"#{node}" [label="#{node} (#{label})"];|
|
|
end
|
|
|
|
defp format_node(:mermaid, node, label) do
|
|
~s|#{node}["#{node} (#{label})"]|
|
|
end
|
|
|
|
defp format_edge(:dot, {in_vert, out_vert}) do
|
|
~s|"#{in_vert}" -> "#{out_vert}";|
|
|
end
|
|
|
|
defp format_edge(:mermaid, {in_vert, out_vert}) do
|
|
"#{in_vert} --> #{out_vert}"
|
|
end
|
|
|
|
defp format_subg(:dot, node, sub_nodes, sub_edges) do
|
|
"""
|
|
subgraph "cluster_#{node}" {
|
|
label="#{node}";
|
|
#{sub_nodes}
|
|
#{sub_edges}
|
|
}
|
|
"""
|
|
end
|
|
|
|
defp format_subg(:mermaid, node, sub_nodes, sub_edges) do
|
|
"""
|
|
subgraph #{node}
|
|
#{sub_nodes}
|
|
#{sub_edges}
|
|
end
|
|
"""
|
|
end
|
|
|
|
# Query Helpers
|
|
|
|
defp workflow_query(wid, deps, opts) do
|
|
query =
|
|
Job
|
|
|> where([j], j.state in @all_states)
|
|
|> order_by(asc: :id)
|
|
|
|
query =
|
|
if opts[:with_subs] do
|
|
where(query, [j], in_workflow_or_sup(j.meta, ^wid))
|
|
else
|
|
where(query, [j], in_workflow(j.meta, ^wid))
|
|
end
|
|
|
|
cond do
|
|
opts[:with_subs] && is_list(opts[:names]) ->
|
|
names = Enum.map(opts[:names], &to_string/1)
|
|
|
|
where(
|
|
query,
|
|
[j],
|
|
fragment("coalesce(?->>'sub_name', ?->>'name')", j.meta, j.meta) in ^names
|
|
)
|
|
|
|
is_list(opts[:names]) ->
|
|
names = Enum.map(opts[:names], &to_string/1)
|
|
|
|
where(query, [j], j.meta["name"] in ^names)
|
|
|
|
opts[:with_subs] && opts[:only_deps] ->
|
|
where(
|
|
query,
|
|
[j],
|
|
fragment("coalesce(?->>'sub_name', ?->>'name')", j.meta, j.meta) in ^deps
|
|
)
|
|
|
|
opts[:only_deps] ->
|
|
where(query, [j], j.meta["name"] in ^deps)
|
|
|
|
true ->
|
|
query
|
|
end
|
|
end
|
|
|
|
defp to_stage_operation({job_id, meta, deps_states}) do
|
|
# A grafted sub-workflow won't have jobs until the grafter runs. Ignore the "deleted" check
|
|
# for the sub-workflow and rely entirely on the grafter job itself.
|
|
deps = Enum.reject(meta["deps"], &match?(["grafted_" <> _, "*"], &1))
|
|
|
|
deps_set =
|
|
if length(deps_states) < length(deps) do
|
|
MapSet.new(["deleted" | deps_states])
|
|
else
|
|
MapSet.new(deps_states)
|
|
end
|
|
|
|
op =
|
|
cond do
|
|
MapSet.member?(deps_set, "deleted") and meta["ignore_deleted"] != true ->
|
|
{:cancelled, :deleted}
|
|
|
|
MapSet.member?(deps_set, "cancelled") and meta["ignore_cancelled"] != true ->
|
|
{:cancelled, :cancelled}
|
|
|
|
MapSet.member?(deps_set, "discarded") and meta["ignore_discarded"] != true ->
|
|
{:cancelled, :discarded}
|
|
|
|
not MapSet.disjoint?(deps_set, @incomplete_set) ->
|
|
:ignore
|
|
|
|
meta["context"] == true ->
|
|
:ignore
|
|
|
|
is_map_key(meta, "orig_scheduled_at") ->
|
|
{:scheduled, DateTime.from_unix!(meta["orig_scheduled_at"], :microsecond)}
|
|
|
|
true ->
|
|
:available
|
|
end
|
|
|
|
{op, job_id}
|
|
end
|
|
|
|
defp apply_operations({:ignore, _}, _), do: []
|
|
|
|
defp apply_operations({op, ids}, conf) do
|
|
now = DateTime.utc_now()
|
|
|
|
base =
|
|
if Utils.has_legacy_workflows?(conf) do
|
|
Job
|
|
|> where([j], j.id in ^ids and j.state in ["scheduled", "suspended"])
|
|
|> update([j], set: [meta: drop_hold(j.meta)])
|
|
else
|
|
where(Job, [j], j.id in ^ids and j.state == "suspended")
|
|
end
|
|
|
|
query =
|
|
case op do
|
|
:available ->
|
|
base
|
|
|> select([j], nil)
|
|
|> update([j], set: [scheduled_at: ^now, state: "available"])
|
|
|
|
{:scheduled, at} ->
|
|
base
|
|
|> select([j], nil)
|
|
|> update([j], set: [scheduled_at: ^at, state: "scheduled"])
|
|
|
|
{:cancelled, reason} ->
|
|
error = %{
|
|
kind: :error,
|
|
reason: Oban.Pro.WorkflowError.exception(reason),
|
|
stacktrace: []
|
|
}
|
|
|
|
formatted = Job.format_attempt(%Job{attempt: 1, unsaved_error: error})
|
|
|
|
base
|
|
|> select([j], j.meta["workflow_id"])
|
|
|> update([j],
|
|
set: [
|
|
cancelled_at: ^now,
|
|
state: "cancelled",
|
|
errors: concat_errors(j.errors, ^[formatted])
|
|
]
|
|
)
|
|
end
|
|
|
|
{_count, workflow_ids} = Repo.update_all(conf, query, set: [])
|
|
|
|
with {:cancelled, reason} <- op do
|
|
Task.start(fn ->
|
|
conf
|
|
|> Repo.all(where(Job, [j], j.id in ^ids))
|
|
|> Enum.each(&Hooks.execute_after_cancelled(reason, %{&1 | conf: conf}))
|
|
end)
|
|
end
|
|
|
|
workflow_ids
|
|
|> Enum.reject(&is_nil/1)
|
|
|> Enum.uniq()
|
|
end
|
|
|
|
# Recorded Helpers
|
|
|
|
defp extract_recorded(meta) do
|
|
case meta do
|
|
%{"recorded" => true, "return" => value, "safe_decode" => true} ->
|
|
Utils.decode64(value, [:safe])
|
|
|
|
%{"recorded" => true, "return" => value} ->
|
|
Utils.decode64(value, [])
|
|
|
|
_ ->
|
|
nil
|
|
end
|
|
end
|
|
end
|