Job Iteration API
Meet Iteration, an extension for ActiveJob that makes your jobs interruptible and resumable, saving all progress that the job has made (aka checkpoint for jobs).
Imagine the following job:
class SimpleJob < ActiveJob::Base def perform User.find_each do |user| user.notify_about_something end end end
The job would run fairly quickly when you only have a hundred
User records. But as the number of records grows, it will take longer for a job to iterate over all Users. Eventually, there will be millions of records to iterate and the job will end up taking hours or even days.
With frequent deploys and worker restarts, it would mean that a job will be either lost of started from the beginning. Some records (especially those in the beginning of the relation) will be processed more than once.
Cloud environments are also unpredictable, and there's no way to guarantee that a single job will have reserved hardware to run for hours and days. What if AWS diagnosed the instance as unhealthy and will restart it in 5 minutes? What if a Kubernetes pod is getting evicted? Again, all job progress will be lost. At Shopify, we also use it to interrupt workloads safely when moving tenants between shards and move shards between regions.
Software that is designed for high availability must be resilient to interruptions that come from the infrastructure. That's exactly what Iteration brings to ActiveJob. It's been developed at Shopify to safely process long-running jobs, in Cloud, and has been working in production since May 2017.
We recommend that you watch one of our conference talks about the ideas and history behind Iteration API.
Add this line to your application's Gemfile:
And then execute:
In the job, include
JobIteration::Iteration module and start describing the job with two methods (
each_iteration) instead of
class NotifyUsersJob < ActiveJob::Base include :: def build_enumerator(cursor:) enumerator_builder.active_record_on_records( User.all, cursor: cursor, ) end def each_iteration(user) user.notify_about_something end end
each_iteration will be called for each
User model in
User.all relation. The relation will be ordered by primary key, exactly like
Check out more examples of Iterations:
class BatchesJob < ActiveJob::Iteration def build_enumerator(product_id, cursor:) enumerator_builder.active_record_on_batches( Product.find(product_id).comments, cursor: cursor, batch_size: 100, ) end def each_iteration(batch_of_comments, product_id) # batch_of_comments will contain batches of 100 records Comment.where(id: batch_of_comments.map(&:id)).update_all(deleted: true) end end
class ArrayJob < ActiveJob::Iteration def build_enumerator(cursor:) enumerator_builder.array(['build', 'enumerator', 'from', 'any', 'array'], cursor: cursor) end def each_iteration(array_element) # use array_element end end
class CsvJob < ActiveJob::Iteration def build_enumerator(import_id, cursor:) import = Import.find(import_id) ::.(import.csv).(cursor: cursor) end def each_iteration(csv_row) # insert csv_row to database end end
Iteration hooks into Sidekiq and Resque out of the box to support graceful interruption. No extra configuration is required.
For more detailed documentation, see rubydoc.
ActiveJob is the primary requirement for Iteration. While there's nothing that prevents it, Iteration is not yet compatible with vanilla Sidekiq API.
Iteration job must respond to
build_enumerator must return Enumerator object that respects the
Unless you are running on Heroku, we recommend you to tune Sidekiq's timeout option from the default 8 seconds to 25-30 seconds, to allow the last
each_iteration to complete and gracefully shutdown.
There a few configuration assumptions that are required for Iteration to work with Resque.
GRACEFUL_TERM must be enabled (giving the job ability to gracefully interrupt), and
FORK_PER_JOB is recommended to be disabled (set to
Why can't I just iterate in
#perform method and do whatever I want? You can, but then your job has to comply with a long list of requirements, such as the ones above. This creates leaky abstractions more easily, when instead we can expose a more powerful abstraction for developers--without exposing the underlying infrastructure.
What happens when my job is interrupted? A checkpoint will be persisted to Redis after the current
each_iteration, and the job will be re-enqueued. Once it's popped off the queue, the worker will work off from the next iteration.
What happens with retries? An interruption of a job does not count as a retry. The iteration of job that caused the job to fail will be retried and progress will continue from there on.
What happens if my iteration takes a long time? We recommend that a single
each_iteration should take no longer than 30 seconds. In the future, this may raise an exception.
Why is it important that
each_iteration takes less than 30 seconds? When the job worker is scheduled for restart or shutdown, it gets a notice to finish remaining unit of work. To guarantee that no progress is lost we need to make sure that
each_iteration completes within a reasonable amount of time.
What do I do if each iteration takes a long time, because it's doing nested operations? If your
each_iteration is complex, we recommend enqueuing another job, which will run your nested business logic. We may expose primitives in the future to do this more effectively, but this is not terribly common today. We recommend to read https://goo.gl/UobaaU to learn more about nested operations.
Why do I use have to use this ugly helper in
build_enumerator? Why can't you automatically infer it? This is how the first version of the API worked. We checked the type of object returned by
build_enumerable, and whether it was ActiveRecord Relation or an Array, we used the matching adapter. This caused opaque type branching in Iteration internals and it didn’t allow developers to craft their own Enumerators and control the cursor value. We made a decision to always return Enumerator instance from
build_enumerator. Now we provide explicit helpers to convert ActiveRecord Relation or an Array to Enumerator, and for more complex iteration flows developers can build their own
What is the difference between Enumerable and Enumerator? We recomend this post to learn more about Enumerators in Ruby.
My job has a complex flow. How do I write my own Enumerator? Iteration API takes care of persisting the cursor (that you may use to calculate an offset) and controlling the job state. The power of Enumerator object is that you can use the cursor in any way you want. One example is a cursorless job that pops records from a datastore until the job is interrupted:
class MyJob < ActiveJob::Base include :: def build_enumerator(cursor:) Enumerator.new do Redis.lpop("mylist") # or: Kafka.poll(timeout: 10.seconds) end end def each_iteration(element_from_redis) # ... end end
This project would not be possible without these individuals (in alphabetical order):
- Daniella Niyonkuru
- Emil Stolarsky
- Florian Weingarten
- Guillaume Malette
- Hormoz Kheradmand
- Mohamed-Adam Chaieb
- Simon Eskildsen
After checking out the repo, run
bundle install to install dependencies. Then, run
bundle exec rake test to run the tests.
To install this gem onto your local machine, run
bundle exec rake install. To release a new version, update the version number in
version.rb, and then run
bundle exec rake release, which will create a git tag for the version, push git commits and tags, and push the
.gem file to rubygems.org.
Bug reports and pull requests are welcome on GitHub at https://github.com/Shopify/job-iteration. This project is intended to be a safe, welcoming space for collaboration, and contributors are expected to adhere to the Contributor Covenant code of conduct.
The gem is available as open source under the terms of the MIT License.
Code of Conduct
Everyone interacting in the Job::Iteration project’s codebases, issue trackers, chat rooms and mailing lists is expected to follow the code of conduct.