Add parallel iteration scaffold

[Mangara]

What and why?

It's sometimes useful to iterate over a collection in parallel to speed up the completion of the work. This PR adds a primitive to enable that:

def build_enumerator(cursor:)
  enumerator_builder.parallel(instances: 7, cursor: cursor) do |instance, total_instances, inner_cursor|
    enumerator_builder.active_record_on_records(
      Product.where("id % ? = ?", total_instances, instance),
      cursor: inner_cursor,
    )
  end
end

When this job is enqueued, it first runs with a nil cursor, causing it to enqueue the 5 child jobs, each with a cursor that looks like { instance: x, inner_cursor: nil }, where x ranges from 0 to 4. When those jobs start, they build the inner enumerator and run as normal, except that the outer enumerator ensures that the cursor stays wrapped in this hash with instance and inner_cursor.

sequenceDiagram
      participant Parent as ParentJob
      participant Q as ActiveJob queue
      participant Child as ChildJob (instance i)

      activate Parent
      Parent->>Parent: build_enumerator(cursor: nil)
      Note over Parent: cursor is nil →<br/>user block NOT invoked
      Parent->>Parent: enqueue_jobs(self.class, arguments)
      Parent->>Q: perform_all_later([job_0..job_N-1])
      Note right of Parent: each child gets<br/>cursor_position =<br/>{instance: i, inner_cursor: nil}
      deactivate Parent

      Q->>Child: perform<br/>(cursor: {instance: i, inner_cursor: nil})
      activate Child
      Child->>Child: build_enumerator(cursor: {instance: i, ...})
      Note over Child: cursor non-nil →<br/>user block IS invoked,<br/>builds inner enum for instance i

      loop until done or interrupted
          Child->>Child: each_iteration(record)
          Note right of Child: cursor_position =<br/>{instance: i, inner_cursor: X}
      end

      alt interrupted
          Child->>Q: retry_job (carries current cursor)
          Q->>Child: perform<br/>(cursor: {instance: i, inner_cursor: X})
          Note over Child: cursor still non-nil →<br/>no re-fan-out,<br/>resumes from inner_cursor X
      else completes
          Note over Child: on_complete callback fires
      end
      deactivate Child

Loading

Gotchas

• Callback behaviour could be unexpected, as the parent job runs no callbacks, and each child job runs the full set. So on_start means this particular instance is starting and on_complete means this particular instance is done iterating. There is no "all child jobs are done iterating" callback, as that would require some form of external synchronization.

Follow-up

• I plan to add parallel versions of most existing enumerators that divvy up the work for you, so you can simply call enumerator_builder.parallel_active_record_on_records(Product.all, instances: 5, cursor: cursor)
• This needs documentation

[CelinaAssal]

Nice!

[adrianna-chang-shopify]

Couple small questions but looks good overall!

[adrianna-chang-shopify]

Hmm, does this mean that if any of the child jobs fail, we re-enqueue all of the children again on retry? I guess we assume idempotency and it's not really an issue?

[Mangara]

Yeah. This is a tricky case where I don't see any perfect options.

I suspect that usually either all the enqueues will fail, or all will succeed. If we do have a partial success, retrying the job will indeed enqueue additional copies.