We’re working with a relatively involved setup, so I’ll try to simplify and keep it brief - details of our config are at the bottom of the post
I’ve built up a Workflow, where the first worker dynamically fans out to N children and adds a single “finisher” worker to handle any final tasks - let’s call them A, B_0…B_n, and Z
All workers are using Oban.Pro.Workers.Workflow, the A worker gets enqueued on its own and inside the process/1 callback appends the rest of the workers to its workflow based on the data it fetches - something like this (simplified code):
@impl true
def process(%Job{args: args} = job) do
b_jobs = generate_b_job_changesets()
workflow = append_workflow(job)
workflow =
Enum.reduce(b_jobs, workflow, fn {name, changeset}, workflow ->
add(workflow, name, changeset, deps: [@recording_extractor_job_name])
end)
b_job_names = Map.keys(chunk_jobs)
workflow =
add(workflow, :z, z_job_changeset(),
deps: b_job_names
)
{:ok, Oban.insert_all(workflow)}
end
The whole setup works as expected in E2E tests, but when run in a pre-production environment, the A worker completes successfully, but the B_x and Z workers fail with the error message of upstream deps deleted
I was hoping it was my mistake - messing up the workflow concatenation or dependency names, but when I fetch the job rows from the DB it all looks correct (some values redacted):
A job row:
id | 350316
state | completed
queue | fact_extraction_initialization
worker | Foo.Bar.A
attempt | 1
max_attempts | 5
priority | 0
args | {"redacted": "redacted"}
meta | {...}
attempted_by | {redacted}
errors | {}
tags | {redacted}
inserted_at | 2024-08-01 18:36:20.930824
scheduled_at | 2024-08-01 18:36:35.930513
attempted_at | 2024-08-01 18:36:36.592045
cancelled_at |
completed_at | 2024-08-01 18:36:36.631846
discarded_at |
One of B job rows:
id | 350317
state | cancelled
queue | fact_extraction
worker | Foo.Bar.B
attempt | 0
max_attempts | 5
priority | 0
args | {"redacted": "redacted"}
meta | {...}
attempted_by |
errors | {"{\"at\": \"2024-08-01T18:36:36.632222+00:00\", \"error\": \"upstream deps deleted\", \"attempt\": 0}"}
tags | {redacted}
inserted_at | 2024-08-01 18:36:36.627618
scheduled_at | 3000-01-01 00:00:00
attempted_at |
cancelled_at | 2024-08-01 18:36:36.632222
completed_at |
discarded_at |
If I’m reading this right the workflow IDs and dependency names align, so it should be working.
Any thoughts what’s happening here? I put a good amount of time implementing the feature with a workflow and I have very limited time to wrap it up right now.
Setup details:
$ mix hex.outdated | grep oban
oban 2.17.12 2.18.0 Update not possible
oban_pro 1.4.10 1.4.10 Up-to-date
oban_web 2.10.4 2.10.5 Update possible
(Most of) oban config:
config :redacted, Oban,
engine: Oban.Pro.Engines.Smart,
repo: Redacted.Repo,
notifier: Oban.Notifiers.PG,
peer: Oban.Peers.Global,
plugins: [
{Oban.Pro.Plugins.DynamicPartitioner,
schedule: if(Mix.env() == :prod, do: "* 1-7 * * *", else: "* * * * *"),
timeout: 60_000,
retention: [completed: 2, discarded: 30]},
{Oban.Plugins.Cron,
crontab: [redacted]},
Oban.Pro.Plugins.DynamicLifeline,
{Oban.Plugins.Reindexer, schedule: "40 4 * * *"}
],
My (uneducated) guess is that because we’re using Oban.Pro.Plugins.DynamicPartitioner, the B_x jobs were getting started when the A job was getting moved from the available/scheduled partition and into the completed partition, which is why it didn’t show up in the query and got interpreted as deleted.
I know if that’s the case I could theoretically use ignore_deleted: true, but that feels like just asking for trouble - failing jobs could also be misinterpreted as deleted when they’re moved to a discarded partition and the workflow would continue even though it shouldn’t - I was counting on stronger guarantees.
Plus IIRC we did introduce the DynamicPartitioner because the impact of oban job churn started to get really noticeable, I don’t think we can really get away from it…
Also, it might or might not be relevant, but the A worker is in a different queue than the B_x and Z workers - I assumed that wasn’t an issue because I couldn’t find anything indicating that in either the docs or the code itself.
