Oban.Pro.Workflow: how to append to workflow that's already running?

Questions & Help Questions
,
1

Hey,

I have a case similar to the video processing example here: Composition — Oban Pro v1.6.4

Composing the steps as Workflow steps work like a charm. However, I’m trying to get one optimization.

In my case, I’m generating AI videos from an input image. Before the actual video can be generated, I need to extract the description of the image using an LLM and then also conditionally do some preprocessing such as cropping.

In other words, there’s the preparation step and the video generation step.

Now, it’s possible that the user generates different videos from the same input image, in which case I don’t want to redo the preparation steps.

If the preparation for that image has already been completed, things are easy – I can just re-use the results.

But it’s possible that the user would start the generation of several videos in a short burst.

So, when e.g. the second request comes in, the preparation for the supplied image has already been started by the first request.

In this case, I’d like to schedule the video generation to when the preparation (that’s already started by the first request) is done.

The current implementation that doesn’t account for this looks something like this

workflow =
  Workflow.new()
  |> Workflow.put_context(%{input: input, video_request_id: request.id})
  |> Workflow.add_workflow(
    :preprocess,
    Workflow.new(workflow_name: "video-preprocess-#{input.input_image_key}")
    |> Workflow.add_cascade(:ensure_extract_subject, &ensure_extract_subject/1)
    |> Workflow.add_cascade(:ensure_outpaint, &ensure_outpaint/1, deps: :ensure_extract_subject, max_attempts: 1)
  )
  |> Workflow.add_cascade(:generate_video, &generate_video/1, deps: :preprocess)
  |> Oban.insert_all()

Question: is there some simplish way to identify that there’s already pre-processing in progress for input.input_image_key and in that case schedule generate_video to the end of it while also overriding the context for that delayed generate_video job?

2

The short answer is no, there’s no simple bit of configuration that will just handle that scenario. You can identify that there’s processing in progress with a query, but you can’t override the context at that point. It is probably best to have the generate_video job check for external context/state itself and act accordingly, rather than manipulate the ongoing workflow.

3

Thanks for your response!

Checking for the state of preprocessing is the easy part. Independently of where you do it, how would you approach “waiting” until the pending preprocessing started earlier is complete?

In the way that also doesn’t interfere with the deployment process (aka not having long-running jobs) and guarantees exact-once execution (aka no PubSub because there’s no guarantee that the listener is there to catch the important event).

Ideally, I want to understand that preprocessing is in progress (again, this is one simple Oban.Job lookup) and schedule my job after that one is done.

I have a WIP solution, but I feel like it’s quite hacky:

First, I look for the existing job like so:

def get_pending_step_of_workflow(opts) do
  workflow_name = Keyword.get(opts, :workflow_name)
  step_name = Keyword.fetch!(opts, :step_name) |> to_string()

  from(oj in Oban.Job,
    where: oj.meta["workflow_name"] == ^workflow_name,
    where: oj.state not in ["completed", "discarded", "cancelled"],
    where: fragment("?->>'name'", oj.meta) == ^step_name,
    limit: 1,
    order_by: [desc: oj.id]
  )
  |> App.Repo.one()
end

If I do find it, I insert a special “no-op” job into the new workflow that has state=scheduled similarly to how Workflow itself works.

That no-op job has meta.start_after_job_id set to the target job.

Code:

Workflow.new()
|> Workflow.put_context(%{input: input, video_request_id: request.id})
|> App.Workflow.start_after_another_job(
  :preprocess,
  existing_workflow_job.id
)
|> Workflow.add_cascade(:generate_video, &App.Img2Video.Generate.generate_video/1,
  deps: :preprocess,
  ignore_discarded: true,
  ignore_cancelled: true
)

Under the hood:

new(%{},
  state: "scheduled",
  scheduled_at: ~U[3000-01-01 00:00:00.000000Z],
  max_attempts: 1,
  meta: %{
    start_after_job_id: job_id,
  }
)

and then, after any job is done (monitored via Telemetry), I make a lookup for that meta.start_after_job_id and start any jobs it found.

job_in_terminal_state? = meta.state == :success or meta.attempt >= meta.max_attempts

if job_in_terminal_state? do
  from(oj in Oban.Job,
    where: fragment("?->>'start_after_job_id'", oj.meta) == ^to_string(meta.job.id),
    where: oj.state == "scheduled"
  )
  |> App.Repo.update_all(
    set: [
      state: "available",
      scheduled_at: DateTime.utc_now()
    ]
  )
end

This kinda works, but I wish it was a part of Workflows somehow. Maybe in the form of a unique key on a sub-workflow that would attempt to re-use an existing one, or something of that sort.

4

One possibility might be to graft the video generation step and dynamically check for pending generation requests. Your workflow could check an externally persisted collection of input_image_keys and add to the collection/exit early, then when the first workflow finishes you can fan out a bunch of image generation steps via graft. Your grafting step would probably need to handle race conditions by ensuring there are no active workflows which could add a new key to the collection. It’s still kinda hacky, but it’s less hacky?