Продолжение Active Job

Продолжения предоставляют механизм прерывания и возобновления задач. Это позволяет длительным задачам продолжать выполнение после перезапуска приложения.

Чтобы включить поддержку продолжений, задачи должны подключать модуль ActiveJob::Continuable. Задачи с поддержкой продолжений автоматически запускаются повторно после прерывания.

Используйте метод step, чтобы определить этапы задачи. Для отслеживания хода выполнения этапа можно использовать необязательный курсор.

Этапы выполняются сразу после обнаружения. Если задача прервана, ранее завершённые этапы будут пропущены. Если этап выполняется, его выполнение возобновится с последнего записанного значения курсора.

Код, не являющийся частью этапа, будет выполняться при каждом запуске задачи.

В метод этапа можно передать блок или имя метода. Блок будет вызван с объектом этапа в качестве аргумента. Методы могут принимать либо ноль аргументов, либо один аргумент — объект этапа.

class ProcessImportJob < ApplicationJob
  include ActiveJob::Continuable

  def perform(import_id)
    # This always runs, even if the job is resumed.
    @import = Import.find(import_id)

    step :validate do
      @import.validate!
    end

    step(:process_records) do |step|
      @import.records.find_each(start: step.cursor) do |record|
        record.process
        step.advance! from: record.id
      end
    end

    step :reprocess_records
    step :finalize
  end

  def reprocess_records(step)
    @import.records.find_each(start: step.cursor) do |record|
      record.reprocess
      step.advance! from: record.id
    end
  end

  def finalize
    @import.finalize!
  end
end

Курсоры

Курсоры используются для отслеживания хода выполнения этапа. Курсором может быть любой объект, который можно сериализовать в качестве аргумента для ActiveJob::Base.serialize. По умолчанию используется nil.

При возобновлении этапа восстанавливается последнее значение курсора. Код этапа отвечает за использование курсора для продолжения выполнения с нужного места.

set! задаёт курсору определённое значение.

step :iterate_items do |step|
  items[step.cursor..].each do |item|
    process(item)
    step.set! (step.cursor || 0) + 1
  end
end

Начальное значение курсора можно задать при определении этапа:

step :iterate_items, start: 0 do |step|
  items[step.cursor..].each do |item|
    process(item)
    step.set! step.cursor + 1
  end
end

Курсор можно переместить вперёд с помощью advance!. Этот метод вызывает succ для текущего значения курсора. Если курсор не реализует succ, возникает исключение ActiveJob::Continuation::UnadvanceableCursorError.

step :iterate_items, start: 0 do |step|
  items[step.cursor..].each do |item|
    process(item)
    step.advance!
  end
end

В advance! можно передать необязательный аргумент from. Это полезно при переборе коллекции записей, идентификаторы которых могут идти не подряд.

step :process_records do |step|
  import.records.find_each(start: step.cursor) do |record|
    record.process
    step.advance! from: record.id
  end
end

Для перебора вложенных записей можно использовать массив:

step :process_nested_records, start: [ 0, 0 ] do |step|
  Account.find_each(start: step.cursor[0]) do |account|
    account.records.find_each(start: step.cursor[1]) do |record|
      record.process
      step.set! [ account.id, record.id + 1 ]
    end
    step.set! [ account.id + 1, 0 ]
  end
end

Установка или перемещение курсора вперёд создаёт контрольную точку. Контрольную точку также можно создать вручную, вызвав метод checkpoint! для этапа. Это полезно, если нужно допустить прерывания, но не требуется обновлять курсор.

step :destroy_records do |step|
  import.records.find_each do |record|
    record.destroy!
    step.checkpoint!
  end
end

Контрольные точки

Контрольная точка — это место, в котором задача может быть прервана. В контрольной точке задача вызывает queue_adapter.stopping?. Если метод возвращает true, задача вызывает исключение ActiveJob::Continuation::Interrupt.

Перед началом каждого этапа, кроме первого этапа каждого запуска задачи, автоматически создаётся контрольная точка. Внутри этапа контрольная точка создаётся при вызове set!, advance! или checkpoint!.

Задачи не прерываются автоматически при переходе адаптера очереди в состояние остановки — они будут выполняться до следующей контрольной точки или до остановки процесса.

Это позволяет прерывать задачи в безопасной точке, однако также означает, что задачи должны создавать контрольные точки чаще, чем истекает время ожидания завершения работы, чтобы перезапуск прошёл корректно.

После прерывания задача автоматически запускается повторно, а данные о ходе выполнения сериализуются в данные задачи под ключом continuation.

Сериализованные данные о ходе выполнения содержат:

• список завершённых этапов

• текущий этап и значение его курсора (если этап выполняется)

Изолированные этапы

Этапы выполняются последовательно в рамках одного запуска задачи, если только задача не прервана.

Чтобы этап всегда выполнялся в отдельном запуске, передайте параметр +isolated: true+.

Это полезно для длительных этапов, в которых невозможно создать контрольную точку в течение периода допустимого выполнения задачи: благодаря этому данные о ходе выполнения сериализуются в данные задачи до начала этапа.

step :quick_step1
step :slow_step, isolated: true
step :quick_step2
step :quick_step3

Ошибки

Если задача вызывает ошибку и не запускается повторно средствами Active Job, ошибка будет передана базовому обработчику очереди, а все данные о ходе выполнения в этом запуске будут потеряны.

Чтобы снизить риск потери данных, задача автоматически запускается повторно, если ошибка возникает после того, как был достигнут прогресс. Прогрессом считается завершение этапа или перемещение курсора вперёд в рамках текущего этапа.

Конфигурация

Для задач с поддержкой продолжений Continuable доступно несколько параметров конфигурации:

• :max_resumptions — максимальное количество возобновлений задачи. По умолчанию используется nil, что означает отсутствие ограничения на количество возобновлений.

• :resume_options — параметры, передаваемые в retry_job при возобновлении задачи. По умолчанию используется { wait: 5.seconds }. Доступные параметры см. в разделе ActiveJob::Exceptions#retry_job.

• :resume_errors_after_advancing — следует ли возобновлять задачу после ошибок, возникших после продвижения выполнения. По умолчанию используется true.