Makes your background jobs interruptible and resumable by design.
Makes your background jobs interruptible and resumable by design.
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 (cursor) { enumerator_builder.active_record_on_records(Shop.all, cursor: cursor) },
->(shop, cursor) { enumerator_builder.active_record_on_records(shop.products, cursor: cursor) },
->(_shop, product, cursor) { enumerator_builder.active_record_on_batch_relations(product.product_variants, cursor: cursor) }
],
cursor: cursor
)
end
def each_iteration(product_variants_relation)
# do something
end
endIteration hooks into most popular queue adapters out of the box to support graceful interruption. No extra configuration is required.
Job-iteration currently supports the following queue adapters (in order of implementation):
It supports the following platforms:
Support for older platforms that have reached end of life may occasionally be dropped if maintaining backwards compatibility is cumbersome.
For more detailed documentation, see rubydoc.
Active Job is the primary requirement for Iteration. For iteration without Active Job in Sidekiq, see Sidekiq Iteration.
Iteration job must respond to build_enumerator and each_iteration methods. build_enumerator must return an Enumerator object that respects the cursor value.
Running iterating jobs on Sidekiq should work with the default configuration. The most important setting is Sidekiq's timeout option, which defaults to 25 seconds. That allows 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 false).
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. If an exception occurs, the job will retry or be discarded as normal using Active Job configuration for the job. If the job retries, it re-processes the iteration that originally failed and progress will continue from there on if successful.
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 runs quickly? 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. The exact timeout depends on your queue adapter configuration.
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 Enumerator objects.
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 < ApplicationJob
include JobIteration::Iteration
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
endThis project would not be possible without these individuals (in alphabetical order):
After checking out the repo, run bin/setup to install dependencies and create mysql database. 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.
Everyone interacting in the Job::Iteration project’s codebases, issue trackers, chat rooms and mailing lists is expected to follow the code of conduct.
No open issues yet, or sync has not completed.