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
