Sidekiq Iteration

Sidekiq Iteration allows you to define jobs that can be safely interrupted and resumed. Instead of a single massive #perform method, you split the job into two parts: an Enumerator (to define the data source and cursor) and an Action (to define what to do with each item).

The Workflow

1. build_enumerator(cursor:): This method is called to construct an Enumerator. It uses the provided cursor to know where to start.
2. Iteration: The job iterates over the items provided by the enumerator.
3. Interruption: If the process receives a SIGTERM (graceful termination signal), the job completes its current iteration, checks job_should_exit?, and then stops.
4. Resumption: Upon stopping, the job re-enqueues itself with the latest cursor_position. When the job runs again, the cursor passed to build_enumerator will reflect the last successfully processed position, allowing the job to pick up exactly where it left off.

class NotifyUsers
  include Sidekiq::Job
  include SidekiqIteration::Iteration

  def build_enumerator(cursor:)
    active_record_records_enumerator(User.all, cursor: cursor)
  end

  def each_iteration(user)
    user.notify_about_something
  end
end

Iteration uses the Ruby Enumerator pattern to allow any resource to be treated as a collection. To create a custom enumerator, you must implement a build_enumerator method in your Sidekiq job that returns an Enumerator object.

An enumerator must yield exactly two positional arguments in this order:

1. The object to be processed: This is passed as the first argument to your job's each_iteration method.
2. A cursor position: This value is persisted by Iteration if the job is interrupted (e.g., during a Sidekiq shutdown or reaching max_job_runtime). This cursor is passed back to build_enumerator as the cursor: keyword argument when the job resumes. The cursor can be any data type that your job backend can serialize and deserialize.

Important Lifecycle Note: Code written after the yield in your enumerator is not guaranteed to execute if the job is forced to exit. If the job is interrupted during a yield, it is re-enqueued, and the remaining code in the enumerator will not run.

def build_enumerator(args, cursor:) 
  Enumerator.new do |yielder| 
    # ... logic ...
    yielder.yield(item, cursor_value)
  end
end

Sidekiq Iteration provides several helper enumerators to handle different data sources:

• Batches of Active Record objects: Use active_record_batches_enumerator to process records in chunks. This is useful for reducing database load.
• Active Record Relations: Use active_record_relations_enumerator to iterate over a relation (like Product.comments).
• Arbitrary Arrays: Use array_enumerator to turn a standard Ruby array into an interruptible iteration.
• CSV Files: Use csv_enumerator to iterate over rows in a CSV file.

# Example: Iterating over batches
class BatchesJob
  include Sidekiq::Job
  include SidekiqIteration::Iteration

  def build_enumerator(product_id, cursor:)
    active_record_batches_enumerator(
      Comment.where(product_id: product_id).select(:id),
      cursor: cursor,
      batch_size: 100,
    )
  end

  def each_iteration(batch_of_comments, product_id)
    comment_ids = batch_of_comments.map(&:id)
    CommentService.call(comment_ids: comment_ids)
  end
end

You can interrupt a job and schedule it for retry if a specific condition is met using the throttle_on method. When the condition block evaluates to true, the job is interrupted and retried after a specified backoff period.

If no backoff is specified, the default is 30 seconds.

Key behaviors:

• Multiple conditions: You can define multiple throttle_on blocks in a single job.
• Inheritance: Throttle conditions are inherited by subclasses. New conditions added in a subclass are appended to the existing ones.

class DeleteAccountsThrottledJob
  include Sidekiq::Job
  include SidekiqIteration::Iteration

  # Throttles the job if the database is unhealthy
  # Retries after 1 minute
  throttle_on(backoff: 1.minute) do
    DatabaseStatus.unhealthy?
  end

  def build_enumerator(cursor:)
    active_record_relations_enumerator(Account.inactive, cursor: cursor)
  end

  def each_iteration(accounts)
    accounts.delete_all
  end
end

While sidekiq-iteration loads records in batches of 100 by default, you can control the granularity of your work.

To allow for more frequent interruptions and safer job execution, prefer using active_record_records_enumerator (which yields individual records) over active_record_batches_enumerator (which yields arrays of records).

When using active_record_records_enumerator, your #each_iteration method should perform the work on a single record rather than iterating over a batch inside the method.

# good: yields individual records, allowing frequent interruptions
class RecordsJob
  include Sidekiq::Job
  include SidekiqIteration::Iteration

  def build_enumerator(product_id, cursor:)
    active_record_records_enumerator(
      Comment.where(product_id: product_id),
      cursor: cursor,
      batch_size: 5,
    )
  end

  def each_iteration(comment, product_id)
    comment.destroy
  end
end