Batch Support - #142
Conversation
|
Could that also support adding jobs to already existing batch? Like SleepyJob enqueuing another job that would also be added to the batch. |
@mbajur Definitely. As long as you're in a job that's part of the batch, adding another job to the batch would work fine. It'd be pretty simple to extend the existing code to handle that - something like this would solve your use-case I think? class SleepyJob < ApplicationJob
queue_as :background
def perform(seconds_to_sleep)
Rails.logger.info "Feeling #{seconds_to_sleep} seconds sleepy..."
sleep seconds_to_sleep
batch.enqueue { AnotherJob.perform_later }
end
endI can update the This would only work safely inside of the job - if you were outside of the job, it's possible the batch would finish before the job gets created. |
|
Yes that would absolutely do the trick for me :) Thank you! |
|
Hi @rosa 👋🏼 Congrats on getting SolidQueue past incubation and under the Rails umbrella officially! I'm sure you've got alot on your plate! Are there any questions I can answer in regards to this PR? I can take the interface/functionality further, but I wanted to discuss it a bit before doing that. If there's anything additional you'd like me to tighten up/try out before discussing it, i'm happy to do so. Also ok to just be on hold and not ready to discuss this further atm. Since it's been a couple months, I figured i'd check in. |
|
Hey @jpcamara, so sorry for the delay and the silence here. I just haven't had the proper time to dedicate to this, and I think this requires and deserves more than a quick look. Thank you so much for putting this together! My instinct with this kind of feature is that they require someone to use them before they're ready. From just looking at the code, I'm not quite sure what kind of edge cases and race conditions could arise here. This is how most of Solid Queue has been built: we've used it in HEY before making it "official" for everyone, seeing how it behaves under certain loads and what kind of problems we encounter. We caught and improved quite a few things that way. We don't have a use case for batches right now, so I'm afraid I won't be able to take this feature to this "production-test" point on my side. Do you see yourself using this in a production setting? |
That makes sense! It was a bit of a chicken and an egg issue for me - I wanted to have batches in SolidQueue before starting to transition some things over, because I have code using Sidekiq Pro batches. But I can start experimenting with it now and report back. I'll continue to work on this PR as well in that case, too. |
|
Thanks for all the hard work here. I'm a big fan of batch jobs so I've been keeping my eye on this PR for a while. Sidekiq Pro and others support the notion of child batches. Like batch jobs, child batches let you work with a higher level abstraction that conceptually simplifies your background work. While implementing that in this PR would be feature creep, I wanted to raise awareness of it in hopes that we design with its extensibility in mind. Thanks again @jpcamara |
hey @dimroc! I couldn't agree more. I used child batches in Sidekiq recently in a project and it highlighted the need to add them to this PR - it's an important feature. I've been putting alot of work into releasing a blog series on ruby concurrency and it's been eating up my free coding-related time, but i'm prioritizing getting back to this soon. Thanks for the feedback! |
c58f11d to
5ba1c27
Compare
I've added child batches to this PR @dimroc |
Added support to this PR for enqueueing within a job using the syntax I suggested: |
|
@jpcamara just an idea here. Could this functionality become its own gem? Like an add-on for solid_queue? |
|
One place I think this could be useful could be rolling back changes. Since the move to SQLite doesn't create jobs to a secondary database until after committing records to the main database, there is no atomic guarantee that both records will commit to both databases, leaving us to need manual rollback logic. Additionally, this reminds me that sidekiq had some atomic writing of batch jobs guarantee, either in pro or enterprise. Same thing might happen if you make a change in your database and call an external API. With SQLite you no longer want to make those changes in your database and make the [long-running] external API call in a transaction. Let's say you have to call stripe twice and want to keep a record of what stage you're at. But something goes wrong with the second stripe call (say the charge is declined) and you need to rollback your database so it's not in an inconsistent state. In this case you can queue the rollback as part of the batch. If memory serves, I think the pattern for this sort of thing is called Sagas, or a DAG such as Elixir has with GenStage. I also found this pattern extremely useful in the past because you can parallelize work more effectively. Say I need to make 1000 remote API calls. Each call should really be in its own job so that it can be retried if it hits the API limits and I need to perform some final job once all 1000 API calls have been made and the batch is complete. |
@mariochavez it's true, it probably could be a gem! Sidekiq has batches as a pro feature, but there is also an open-source gem that mostly supports the same api https://github.com/breamware/sidekiq-batch. There are some gotchas with approaching it that way (one, for instance, around jobs being automatically cleaned up and not being able to do anything but warn users against it). But my main motivation is that I personally want it as a first-class feature of SolidQueue. It's a first-class (albeit paid) feature of Sidekiq, and it's a first-class feature of GoodJob. Being built-in means it's more likely to get use/support and alleviates concerns it may be abandoned at some point. I also think it's a great core feature of a job library. Gush is awesome! It definitely works similarly, though this being backed by a DB in SolidQueue means it has more ACID-type guarantees. Something I would like to see is an even more sophisticated "workflow" type layer that worked with any activejob system, and that's something I've toyed around wtih over the past year. That kind of system I think goes a step beyond, is more complicated, and is better served as a separate gem. I think batch support being included in the job server is a good fit. |
|
Thanks for all this work @jpcamara! I'm curious what the expected behavior is with An argument for keeping this in the main gem is to ease integration testing with the other features to increase cohesion and stability. I can see it getting hairy when you stack a few different configurations on top of nested batches, and having to assert proper behavior. https://github.com/rails/solid_queue?tab=readme-ov-file#concurrency-controls |
|
I would love to see batches in Solid Queue! My use case is primarily creating workflows. |
|
Just wanna thank @jpcamara for the work he's done, is doing here. Would love to use this 🙏 |
Hey @kaka-ruto, I think I saw a message about you being willing to try this out? That would be great! Most of my free time is working on a RubyConf talk I have in a couple weeks, but i'll be shifting back to this right after that and will give you an update. |
1ef7f38 to
8099030
Compare
| self.maintenance_queue_name = "default" | ||
|
|
||
| def enqueue(&block) | ||
| raise "You cannot enqueue a batch that is already finished" if finished? |
There was a problem hiding this comment.
Perhaps use here a new error class, like SolidQueue::BatchAlreadyFinished? Or something like that.
rosa
left a comment
rosa
left a comment
There was a problem hiding this comment.
I've done a first pass over the first few important classes, but will continue tomorrow and during the week!
|
|
||
| def check_completion! | ||
| return if finished? || !ready? | ||
| return if batch_executions.limit(1).exists? |
There was a problem hiding this comment.
Why the limit(1).exists? here vs. batch_executions.any? for example?
| (super || {}).with_indifferent_access | ||
| end | ||
|
|
||
| def check_completion! |
There was a problem hiding this comment.
I think I'd name this just check_completion, without the !, to follow the convention that bang-methods have a counterpart without bang that perform the same action but without raising errors.
There was a problem hiding this comment.
Ahh, I see it calls update!, maybe that's why you added the !... hmmm... even with that, since there's no counterpart that doesn't raise, I'd remove the !.
| end | ||
|
|
||
| def ready? | ||
| enqueued_at.present? |
There was a problem hiding this comment.
Something a bit confusing with the status here is that ready? means there's enqueued_at but #status returns processing for that case 🤔 I think ready for me suggests not yet processing, but ready to be processed or ready to start being processed I think... Maybe we could forget about ready and processing here and use something that refers directly to the enqueued_at attribute: enqueued?, and status could also return enqueued.
| end | ||
|
|
||
| def check_completion! | ||
| return if finished? || !ready? |
There was a problem hiding this comment.
Under which circumstance could this be called for a batch that's not ready? (i.e., for which start_batch hasn't been called?)? Trying to figure out if we could remove that condition from here.
| finished_attributes[:completed_jobs] = total_jobs - failed | ||
|
|
||
| update!(finished_attributes) | ||
| execute_callbacks |
There was a problem hiding this comment.
I'd like to rewrite this method to be more clear and to make it perfectly obvious what it does. But then I realise I'm not quite sure what the two main checks here do, before we update the batch. I mean these two checks:
return if batch_executions.limit(1).exists? rows = Batch.where(id: id).unfinished.empty_executions.update_all(finished_at: Time.current)
return if rows.zero?Could you perhaps explain these again? 🙏🏻
|
|
||
| ```rb | ||
| Rails.application.config.after_initialize do # or to_prepare | ||
| SolidQueue::Batch.maintenance_queue_name = "my_batch_queue" |
There was a problem hiding this comment.
Maybe we don't need a custom attribute for this 🤔 Since we're setting it in the initialiser, we can also set it like this:
SolidQueue::Batch::EmptyJob.queue_as "my_batch_queue"
And we can delete the maintenance_queue_name everywhere in Batch.
| serialize :on_finish, coder: JSON | ||
| serialize :on_success, coder: JSON | ||
| serialize :on_failure, coder: JSON |
There was a problem hiding this comment.
I think for these 3, I'd remove some repetition here and below:
%w[ finish success failure ].each do |callback_type|
serialize "on_#{callback_type}", coder: JSON
define_method("on_#{callback_type}=") do |callback|
super serialize_callback(callback)
end
end| active_job = ActiveJob::Base.deserialize(send(job_field)) | ||
| active_job.send(:deserialize_arguments_if_needed) | ||
| active_job.arguments = [ self ] + Array.wrap(active_job.arguments) | ||
| SolidQueue::Job.enqueue_all([ active_job ]) |
There was a problem hiding this comment.
Oh! Any reason we can't just use enqueue here, so we don't have to go and find the job ID afterwards?
| SolidQueue::Job.enqueue_all([ active_job ]) | ||
|
|
||
| active_job.provider_job_id = Job.find_by(active_job_id: active_job.job_id).id | ||
| attrs[job_field] = active_job.serialize |
There was a problem hiding this comment.
We don't do anything with the attrs later, right? But just wondering if I'm missing something.
| update!(enqueued_at: Time.current) | ||
| end | ||
|
|
||
| class << self |
There was a problem hiding this comment.
I'd move this to the beginning of the file, just to be consistent with other classes in the gem (nothing wrong with having it in the end, it's just for consistency 😅).
|
Additionally, I think we could simplify some of the tests by reusing existing dummy jobs like |
|
Thank you so much for the feedback everyone, and any test cleanup would be greatly appreciated @p-schlickmann! I'm finishing up a talk i'm giving at SFRuby conf next week (you should go if you're in the area 😉 https://sfruby.com/), but I will dig into this when i'm back! Thanks again! |
|
Hey @ollym! You bet, this, together with proper sharding support, is my main (only?) goal for Solid Queue in 2026. No official roadmap or plan because the work is pretty much happening on people's free time, but it is definitely happening. |
6cf2ba1 to
fcae152
Compare
|
Hey all - took me a bit longer to get back to this. But I'm back, and excited to get this to the finish line! Here's the current status, and steps to get this all wrapped!
|
|
One thing I forgot, I'm also planning to change the callback method definition - i got some feedback that it's currently confusing, and I can see that: SolidQueue::Batch.enqueue(on_success: SuccessCallbackJob.new(item)...
class SuccessCallbackJob < ApplicationJob
def perform(batch, item) # where did batch come from??
# ...
end
endSo I think I will change it to make batch an available method on the callback, like it is on the regular jobs: SolidQueue::Batch.enqueue(on_success: SuccessCallbackJob.new(item)...
class SuccessCallbackJob < ApplicationJob
def perform(item)
batch.metadata # access `batch` using the method instead
end
end |
|
Ok, I'm finally back at this as well! Thanks a lot for the last push, @jpcamara! 🙏🏻 I'm planning Solid Queue 2.0 with the idea that batches are there.
Yes, this seems reasonable to me! Similar to what we do for other cases, like a job failing to unblock the next one for concurrency controls.
Good point. I think it's ok to change that, especially considering we'll change a major version number. I'm going to do another pass. Thanks a lot! |
Batches group jobs so the progress of the whole set can be tracked and callbacks fired based on its status: on_finish once every job has finished, even when some failed; on_success only when all of them succeeded; and on_failure when at least one failed for good. Callbacks can be given as job classes or configured job instances, they're serialized when the batch is created, and they always enqueue through Solid Queue in the batch's finishing transaction, regardless of the job classes' adapters. A job joins the batch that's active when its enqueue is requested, which also covers enqueues Rails defers until after the surrounding transaction commits, and jobs can enqueue more jobs into their own batch while it runs. Membership lives on the job's batch_id, while a solid_queue_batch_executions row tracks each outstanding attempt: the batch finishes, through a single-winner update, when no tracking rows remain, without ever locking the batch row outside that one moment. Batches that can't detect their own completion — bulk-discarded jobs, processes that crashed between enqueueing and starting a batch, or completions rolled back by failing callback enqueues — are swept by the dispatcher as part of its regular maintenance. Batches track total, completed, failed and pending job counts, take a description and arbitrary metadata, and finished ones are cleared with clear_finished_in_batches, keeping failed batches around for inspection.
* Count logical jobs in batch counters instead of attempts: retries keep their active_job_id, so a job that fails twice and then succeeds contributes 1 to total_jobs, not 3, matching Sidekiq Pro and GoodJob. Every attempt still gets its own tracking row, and the completion machinery is untouched. * Ship the batches schema as an optional migration until 2.0: `rails solid_queue:update` copies a migration guarded with if_not_exists throughout, so fresh installs (which get the tables with the base schema) can run it as a no-op. Until an app migrates, jobs run without batch bookkeeping, starting a batch raises with instructions, and the dispatcher logs a deprecation warning once per process. Includes a documented knob for building the jobs index concurrently on large PostgreSQL tables. * Finish empty batches at start instead of enqueueing an EmptyJob: starting a batch already ends by checking completion, so a batch with no jobs takes the same exit a fast-draining batch takes — callbacks fire right away, total_jobs stays honestly at zero, and no worker is needed. * Rebuild BatchExecution on top of Execution, reusing the base insert machinery, and reorganize Batch into Callbacks, Status, Clearable and Sweepable concerns around a start/finish/finalize lifecycle. * Split the batch sweep into named phases with self-describing metrics: stale_executions, finished_batches and started_batches, instead of repaired/size/started, which said nothing about what was counted. * Accept an explicit metadata: hash in Batch.enqueue, merging any extra keyword arguments into it. * Document why the batch completion re-check holds on every database: PostgreSQL's READ COMMITTED can let a blocked completion update win from a stale subquery snapshot — the re-check catches it with a fresh one — while MySQL reads DML subqueries from the latest committed data, so it can't win wrongly in the first place. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Dropping and re-adding the jobs table's batch_id invalidates cached prepared statements whose SQL text didn't change, and PostgreSQL raises PreparedStatementCacheExpired when one is reused inside a transaction, where the adapter can't silently replan. Which statements are cached depends on which tests ran first, so this only failed on some CI seeds. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
17 participants
Support for batches in SolidQueue!
Batches are a powerful feature in Sidekiq Pro and GoodJob which help with job coordination. A "Batch" is a collection of jobs, and when those jobs meet certain completion criteria it can optionally trigger another job with the batch record as an argument.
The goal of this feature is to:
This PR provides a functional batch implementation. The following scenarios will work:
You should see the following in your logs:
Here is the full interface, demonstrating a few different options:
Some have mentioned the GoodJob example for complex batches and asked if this could be implemented in the SolidQueue::Batch approach: https://github.com/bensheldon/good_job?tab=readme-ov-file#complex-batches. GoodJob offers mutable batches, and the SolidQueue::Batch implementation mostly does not. So this is how you would implement the same, more complex example:
Here are the things that are open questions and missing implementation details:
JobBatchthe right name? General feedback on naming in the featureon_success,on_finish,on_failurediscard_onby marking the job as finished. That means the batch cannot identify that the job actually failed.