felix-starman
I have a reproduction below that I asked Claude to extract from our implementation, and seems to suffer the same issue.
The background is we have a workflow that loads data from our DB (populated from a previous job), and then for each item (for us it’s insurance plans) we need to fetch n child items (policies for the plan) from an API. There are ~200 plans and per plan there’s 300-2000 policies. Due to the API’s pagination, and the variability of the number of policies, we use a sub-workflow per plan.
What I’m seeing is that the sub-workflows never run, or insert into the database.
From the documentation, it seems like I should not be using Oban.insert_all in the sub-workflow defining function, as that would result in orphaned sub-workflows.
Am I misusing apply_graft?
defmodule NestedWorkflowRepro do
@moduledoc """
Minimal reproduction of multi-level nested workflow pattern using Oban.Pro.Workflow.
This emulates the structure in PolicyBotPolicySync:
- Level 1: Main workflow fetches items
- Level 2: Graft per-item workflows
- Level 3: Graft per-chunk workflows within each item workflow
Each step returns simple concatenated strings for easy debugging.
"""
use Oban.Pro.Worker, queue: :default, max_attempts: 1
alias Oban.Pro.Workflow
require Logger
# Level 1: Main workflow functions
@doc """
Fetches the initial list of items to process.
Returns %{items: ["A", "B", "C"]}
"""
def fetch_items(_context) do
items = ["A", "B", "C"]
Logger.info("fetch_items: #{inspect(items)}")
%{items: items, result: "fetched_items"}
end
@doc """
Grafts a sub-workflow for each item.
This is Level 2 grafting - creates per-item workflows.
"""
def graft_per_item(%{fetch_items: %{items: items}}) do
Logger.info("graft_per_item: creating workflows for #{length(items)} items")
{items, &process_item_workflow/2}
|> Workflow.apply_graft(context: %{batch_size: 2})
|> Oban.insert_all()
end
# Level 2: Per-item workflow functions
@doc """
Creates a workflow to process a single item.
This workflow will:
1. Process initial batch
2. Determine if more batches are needed
3. If needed, graft sub-workflows for remaining batches (Level 3)
"""
def process_item_workflow(item, %{batch_size: batch_size}) do
workflow_name = "nested_workflow_repro_item_#{item}"
Workflow.new(workflow_name: workflow_name)
|> Workflow.put_context(%{
item: item,
batch_size: batch_size
})
|> Workflow.add_cascade(:process_initial_batch, &process_initial_batch/1)
|> Workflow.add_cascade(:determine_remaining, &determine_remaining/1, deps: :process_initial_batch)
|> Workflow.add_graft(:process_remaining_batches, &graft_remaining_batches/1,
deps: [:process_initial_batch, :determine_remaining]
)
end
@doc """
Processes the initial batch for an item.
Simulates work that determines total work needed.
"""
def process_initial_batch(%{item: item, batch_size: batch_size}) do
# Simulate: item "A" has 5 total chunks, "B" has 3, "C" has 1
total_chunks =
case item do
"A" -> 5
"B" -> 3
"C" -> 1
end
processed = min(batch_size, total_chunks)
result = "item_#{item}_initial_batch_#{processed}"
Logger.info("process_initial_batch: #{result}")
%{
result: result,
total_chunks: total_chunks,
processed: processed,
needs_more: processed < total_chunks
}
end
@doc """
Determines what remaining work is needed.
"""
def determine_remaining(%{
process_initial_batch: %{total_chunks: total, processed: processed, needs_more: needs_more},
item: item
}) do
result = "item_#{item}_determined_#{total - processed}_remaining"
Logger.info("determine_remaining: #{result}")
%{
result: result,
remaining_chunks: total - processed,
needs_more: needs_more
}
end
@doc """
Grafts sub-workflows for remaining batches if needed.
This is Level 3 grafting - creates per-chunk workflows.
"""
def graft_remaining_batches(%{
process_initial_batch: %{processed: processed, total_chunks: total, needs_more: true},
item: item,
batch_size: batch_size
}) do
# Calculate chunks needed (similar to calculate_policy_chunks)
chunks =
Range.new(processed, total - 1, batch_size)
|> Enum.map(fn idx -> %{chunk_index: idx, size: batch_size} end)
Logger.info("graft_remaining_batches: item=#{item}, chunks=#{length(chunks)}")
# Level 3: Graft per-chunk workflows
{chunks, &process_chunk/2}
|> Workflow.apply_graft()
|> Oban.insert_all()
end
# No additional chunks needed
def graft_remaining_batches(%{process_initial_batch: %{needs_more: false}, item: item}) do
Logger.info("graft_remaining_batches: item=#{item}, no additional chunks needed")
# Return empty workflow
Workflow.new()
|> Oban.insert_all()
end
# Level 3: Per-chunk workflow functions
@doc """
Processes a single chunk within an item's workflow.
"""
def process_chunk(%{chunk_index: idx, size: size}, %{item: item}) do
result = "item_#{item}_chunk_#{idx}_size_#{size}"
Logger.info("process_chunk: #{result}")
%{result: result}
end
@doc """
Builds the main workflow.
This creates a 3-level nested workflow structure:
- Level 1: Main workflow (fetch items -> graft per-item)
- Level 2: Per-item workflows (process initial -> graft per-chunk)
- Level 3: Per-chunk workflows (process individual chunks)
"""
def workflow(opts \\ []) do
batch_size = Keyword.get(opts, :batch_size, 2)
Workflow.new(workflow_name: "nested_workflow_repro_main")
|> Workflow.put_context(%{batch_size: batch_size})
|> Workflow.add_cascade(:fetch_items, &fetch_items/1)
|> Workflow.add_graft(:process_items, &graft_per_item/1, deps: [:fetch_items])
end
@impl Oban.Pro.Worker
def process(%Oban.Job{args: args}) do
batch_size = Map.get(args, "batch_size", 2)
workflow(batch_size: batch_size)
|> Oban.insert_all()
:ok
end
end
Trending in Questions
Other Trending Topics
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #library
- #deployment
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #elixirconf
- #channels
- #exunit
- #discussion
- #code-sync
- #podcasts
- #javascript
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixirconf-us
- #ai
- #blog-post
- #elixir-ls
- #phoenix_html
- #iex
- #graphql
- #genstage
- #websockets
- #supervisor
- #advent-of-code
- #distillery
- #processes
- #api
- #forms
- #metaprogramming
- #hex
- #security











Showing Posts 1 to 9- Show Best Posts
- Show All (oldest first)
- Show All (newest first)
felix-starman
Semi-related, but when then viewing the Cascade jobs in Oban.Web for the
process_item_workflowI get an error.Presumably it’s not able to render the Workflow struct
felix-starman
Changing
graft_per_itemto this:results in an Ecto.MultipleResultsError from trying to fetch multiple contexts…
sorentwo
You need to use
insert_allafterapply_graftor the jobs won’t be inserted at all.This is because the term (whatever the struct name is) isn’t available on the Web node and it’s an unsafe action to render it. You can override that with a custom resolver ( Oban.Web.Resolver — Oban Web v2.12.5 ). It can also happen in development mode because modules are loaded lazily.
That effectively grafts multiple workflows like they are one, which causes that
MultipleResultsErroras you saw.We’re working through your top example to see if there’s an alternate version that will work, or if there’s a bug to fix.
felix-starman
Ok, so at least “surface-level” the place where I’m calling
Oban.insert_allingraft_per_itemmakes sense, since it’s after theapply_graftfelix-starman
Ok <3
I’ll also post here if I find a workaround. I am going to try breaking it into Worker modules at different layers to see if anything helps there too. I also changed it to explicitly set
workflow_idin theWorkflow.newbut it looks like it does the same thing you were mentioning (trying to run it as a single workflow), by using the same workflow as when you don’t explicitly provide itfelix-starman
TL;DR - after refactoring some to work around it, I can’t get
after_cancelledto be called in test for a step dependent on a failing grafted step, but it calls it correctly in dev.I’ve refactored the
process_item_workflowinto its own worker that for right now just builds the workflow then inserts it, so they are independent of the “top-level”/original workflow. I’ve also extracted all theadd_cascadesteps from that “per-item” workflow into small workers, and swapped to justadd.I’ve then added a final step in the per-item workflow that depends on
:process_remaining_batchesand updates the DB to say that it’s all complete, and anafter_cancelledthat marks it as “failed”.This works okay for now, but there’s an issue I’m encountering with testing the
MarkCompletedWorker.after_cancelledscenario. When running manually in dev it correctly fires theprocessif everything runs correctly, and fires theafter_cancelledif one of the grafted steps fails enough.But when I use
run_workflowordrain_jobsI can’t get the callback to fire for the grafted jobs. It does fire the callback when I cause the:process_initial_batchstep’s worker to fail and I usedrain_jobs.run_workflowjust fails from the induced error, but I assume this is intended, since it’s running synchronously.I thought this might be semi-related, since it seems to be around graft+sub-workflow handling.
sorentwo
Nice to hear that you were able to refactor around the issue for now.
Are you calling
run_workflowordrain_jobswith safety enabled (with_safety: true)? If not, a crash or exception will bubble up into the test and theafter_cancelledcallback won’t have a chance to run.felix-starman
Unfortunately, yes that’s using
drain_jobs(queue: :all, with_safety: true)(and same options when testingrun_workflow)Interesting thing I think I forgot to mention is that since I’m tagging
capture_log: trueI can see that it’s logging the (intentionally induced) errors I emit from the grafted jobs when I usedrain_jobs(because ExUnit.CaptureLog only shows the logs if the test fails) but I don’t get any logs if I userun_workflow.This doesn’t explain why the
after_cancelledisn’t running fordrain_jobson the “grafted errors” scenario but it would seem related to whyrun_workflowwasn’t running the callback on the “initial step error” scenario.felix-starman
@sorentwo I saw 1.6.8 came out recently and mentions fixes for nested graphs
I haven’t had a chance to test it yet, but I will try to poke this soon w/ 1.6.8