Grouping support?

[mbajur]

Hello!

I was wondering was there ever a discussion about identical jobs grouping? I'm talking about a case when same jobs are being enqueued with different arguments - it would be great to have an option to group them and run one grouped job instead of X separate jobs. Something like: https://github.com/gzigzigzeo/sidekiq-grouping or https://github.com/dtaniwaki/sidekiq-merger

This feature sounds like something someone probably already requested or asked about but it can be named in a ton of ways so I'm not eve sure how to construct a search query to find it :) Any tips much appreciated!

[bensheldon]

Interesting! Can you give me an example of what you would want to group?

I feel like some combination of job-iteration, and/or batching would get you sorta close, but I'm sorta hazy on the use-case.

Generally, things that are at time-of-enqueue (rather than time-of-perform) are much easier to handle outside of the adapter. Though GoodJob has gotten somewhat feature-rich, the primary goal is still to queue and dequeue jobs as fast as possible.

And at least for Sidekiq, putting data temporarily in Redis is maybe faster than the relational database. But for GoodJob, the argument is usually "why wouldn't this make sense modeled concretely in your application compared to generically in GoodJob."

None of which is to say I'm against it, but more curious with context.

[mbajur]

I would say the simplest example would be webhooks sending - if there is one job per one webhook event (WebhookDispatchJob) and app will produce 100 of such in the span of, say, 1 minute - it would be great to be able to merge all these 100 jobs into one and send them all as one request instead of 100 separate requests. On the other hand, jobs are simply database records so that logic could be happening outside of good_job. So, here is the logic I'm talking about:

1. WebhookDispatchJob.sed(delay: 1.minute).perform_later
2. Check if there are any WebhookDispatchJob jobs already enqueued in last 1 minute
   a. If there are but for single dispatch, update the already existing job record and convert it's hash of arguments to array of hashes (merge all the argument hashes into one array of hashes)
   b. If there are and already for multiple webhooks, update existing job by appending given arguments to the array of hashes
   c. if there are none, create a new record like you normally would

There are two tricky parts in here:

1. making job have either regular arguments defined in perform OR array of hashes sounds a bit sketchy and that would need to be covered in the job logic, i would maybe consider adding another argument and operating on it or using a separate job class or subclass for merged jobs
2. How to actually do the upsert logic in a proper way to not make the enqueueing system slower and to not mess with the job processor? I guess some sort of locking mechanism should be involved as i can see some race condition potential in here

---

I'm writing all that knowing that logic for batching webhooks might be implemented outside of good_jobs scope in thousand of ways that do not involve doing anything on good_job side of things. But let's just consider this as an example and this entire idea as something that could be super convenient to have in order to not implement all that from zero every time one needs it ;)

[bensheldon]

Thanks for sharing all of that. I think I'm skeptical that the bookkeeping necessary to group together multiple webhooks into a single job would be any more efficient than processing them as separate jobs.

My most naive implementation would look something like this:

class WebhookGroupJob
  def self.count_since
    result = WebhookRecord.select("min(created_at) as oldest_at", "count(*) as count").take(:created_at, :count)[0]
    [result[:oldest_at], result[:count]]
  end

  before_enqueue do
    oldest_at, count = count_since
    if count < 49 || (oldest_at && oldest_at < 30.seconds.ago) || arguments.last[:force]
      # include GoodJob::AdvisoryLockable
      WebhookRecord.with_advisory_lock("WebhookGroupJob-before_enqueue_count") do # <-- I need to improve `with_advisory_lock` to allow pg_try_locks to skip the block if already locked
        records = WebhookRecord.all
        WebhookRecord.delete_all(id: records.map(&:id))
        @arguments = [records]
      end
    else
      throw :abort
    end
  end

  def perform(webhooks = [], force: true)
    # ...
  end
end

This does it at enqueue time, which maybe isn't ideal because you want to return a webhook response as quickly as possible.

This would also require having a belt-and-suspenders of a cron job running every 1 minute, for example, to cover the cases where there aren't a lot of webhooks triggering it, because it only allows an enqueue to happen on the event, rather than enqueuing the job

But just thinking through it, that looks like the basic recipe that would be necessary no matter what:

• a min(timestamp)/count query
• some kind of lock around the bookkeeping
• the bookkeeping operation to group and modify/destroy

The challenge of this is sort of like "when does it happen?"

• At enqueue time is the simplest. But that has the downside of adding latency where you most don't want it. It also can really only prevent the enqueue.
• While the job is in queue. I think this would be most brittle/dangerous to attempt to lock the job in a way that would prevent it from being dequeue while being modified, but not actually dequeuing it. But maybe?
• At perform time, which would be similar to enqueue time, but would again involve some locks which would slow down the queue.

It's a hard problem 🤔

[bensheldon]

I'll offer too that maybe this isn't the right tool for a Job system (of discrete tasks) and more something like Stream Processing. It would be a lot easier to do this batching and grouping in memory in realtime 🤔