Active Job Continuations (rails#55127)

Continuations provide a mechanism for interrupting and resuming jobs.
This allows long running jobs to make progress across application
restarts.

Jobs should include the `ActiveJob::Continuable` module to enable
continuations. Continuable jobs are automatically retried when
interrupted.

Use the `step` method to define the steps in your job. Steps can use an
optional cursor to track progress in the step.

Steps are executed as soon as they are encountered. If a job is
interrupted, previously completed steps will be skipped. If a step is in
progress, it will be resumed with the last recorded cursor.

Code that is not part of a step will be executed on each job execution.

You can pass a block or a method name to the step method. The block
will be called with the step object as an argument. Methods can either
take no arguments or a single argument for the step object.

```ruby
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)
        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)
      record.reprocess
      step.advance! from: record.id
    end
  end

  def finalize
    @import.finalize!
  end
end
```

**Cursors**

Cursors are used to track progress within a step. The cursor can be any
object that is serializable as an argument to
`ActiveJob::Base.serialize`. It defaults to `nil`.

When a step is resumed, the last cursor value is restored. The code in
the step is responsible for using the cursor to continue from the right
point.

`set!` sets the cursor to a specific value.

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

A starting value for the cursor can be set when defining the step:

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

The cursor can be advanced with `advance!`. This calls `succ` on
the current cursor value. It raises an
`ActiveJob::Continuation::UnadvanceableCursorError` if the cursor does
not implement `succ`.

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

You can optionally pass a `from` argument to `advance!`. This is
useful when iterating over a collection of records where IDs may not be
contiguous.

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

You can use an array to iterate over nested records:

```ruby
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
```

Setting or advancing the cursor creates a checkpoint. You can also
create a checkpoint manually by calling the `checkpoint!` method on the
step. This is useful if you want to allow interruptions, but don't need
to update the cursor.

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

**Checkpoints**

A checkpoint is where a job can be interrupted. At a checkpoint the job
will call `queue_adapter.stopping?`. If it returns true, the job will
raise an `ActiveJob::Continuation::Interrupt` exception.

There is an automatic checkpoint at the end of each step. Within a step
calling one is created when calling `set!`, `advance!` or
`checkpoint!`.

Jobs are not automatically interrupted when the queue adapter is marked
as stopping - they will continue to run either until the next
checkpoint, or when the process is stopped.

This is to allow jobs to be interrupted at a safe point, but it also
means that the jobs should checkpoint more frequently than the shutdown
timeout to ensure a graceful restart.

When interrupted, the job will automatically retry with the progress
serialized in the job data under the `continuation` key.

The serialized progress contains:
- a list of the completed steps
- the current step and its cursor value (if one is in progress)

**Errors**

If a job raises an error and is not retried via ActiveJob, it will be
passed back to the queue adapter and any progress in this execution will
be lost.

To mitigate this, the job will automatically retried if it raises an
error after it has made progress. Making progress is defined as having
completed a step or advanced the cursor within the current step.

**Queue Adapter support**

Active Job Continuations call the `stopping?` method on the queue
adapter to check if we are in the shutdown phase. By default this will
return false, so the adapters will need to be updated to implement this
method.

This implements the `stopping?` method in the test and Sidekiq adapters.

It would be possible to add support to Delayed Job via a plugin, but it
would probably be better to add a new lifecycle callback to DJ for when
it is shutting down.

Resque also will require a new hook before it can be supported.

Solid Queue's adapter is not part of Rails, but support can be added
there via the `on_worker_stop` hook.

**Inspiration**

This took a lot inspiration from Shopify's
[job-iteration](https://github.com/Shopify/job-iteration) gem.

The main differences are:
- Continuations are Active Job only, so they don't provide the custom
  enumerators that job-iteration does.
- They allow multi-step flows
- They don't intercept the perform method
- Continuations are a sharp knife - you need to manually checkpoint and
  update the cursor. But you could build a job-iteration-like API on top
  of them.

**Future work**

It would be a good exercise to see if the job-iteration gem could be
adapted to run on top of Active Job Continuations to highlight any
missing features - we'd want to add things like max iteration time, max
job runtime and forcing a job to stop.

Another thing to consider is a mechanism for checking whether it is safe
to call a checkpoint. Ideally you wouldn't allow them within database
transactions as they'll cause a rollback. We can maybe inject checkpoint
safety handlers and add a default one that checks whether we are in any
active transactions.

Author: djmb

activejob/CHANGELOG.md

@@ -1,3 +1,42 @@
+*   Allow jobs to the interrupted and resumed with Continuations
+
+    A job can use Continuations by including the `ActiveJob::Continuable`
+    concern. Continuations split jobs into steps. When the queuing system
+    is shutting down jobs can be interrupted and their progress saved.
+
+    ```ruby
+    class ProcessImportJob
+      include ActiveJob::Continuable
+
+      def perform(import_id)
+        @import = Import.find(import_id)
+
+        # block format
+        step :initialize do
+          @import.initialize
+        end
+
+        # step with cursor, the cursor is saved when the job is interrupted
+        step :process do |step|
+          @import.records.find_each(start: step.cursor) do |record|
+            record.process
+            step.advance! from: record.id
+          end
+        end
+
+        # method format
+        step :finalize
+
+        private
+          def finalize
+            @import.finalize
+          end
+      end
+    end
+    ```
+
+    *Donal McBreen*
+
 *   Defer invocation of ActiveJob enqueue callbacks until after commit when
     `enqueue_after_transaction_commit` is enabled.
 

activejob/README.md

@@ -95,6 +95,10 @@ their gem, or as a stand-alone gem. For discussion about this see the
 following PRs: [23311](https://github.com/rails/rails/issues/23311#issuecomment-176275718),
 [21406](https://github.com/rails/rails/pull/21406#issuecomment-138813484), and [#32285](https://github.com/rails/rails/pull/32285).
 
+## Continuations
+
+Continuations allow jobs to be interrupted and resumed. See more at ActiveJob::Continuation.
+
 
 ## Download and installation
 

activejob/lib/active_job.rb

@@ -41,6 +41,7 @@ module ActiveJob
   autoload :SerializationError, "active_job/arguments"
   autoload :UnknownJobClassError, "active_job/core"
   autoload :EnqueueAfterTransactionCommit
+  autoload :Continuation
 
   eager_autoload do
     autoload :Serializers

activejob/lib/active_job/continuable.rb

@@ -0,0 +1,59 @@
+# frozen_string_literal: true
+
+module ActiveJob
+  # = Active Job Continuable
+  #
+  # Mix ActiveJob::Continuable into your job to enable continuations.
+  #
+  # See +ActiveJob::Continuation+ for usage. # The Continuable module provides the ability to track the progress of your jobs,
+  # and continue from where they left off if interrupted.
+  #
+  module Continuable
+    extend ActiveSupport::Concern
+
+    CONTINUATION_KEY = "continuation"
+
+    included do
+      retry_on Continuation::Interrupt, attempts: :unlimited
+      retry_on Continuation::AfterAdvancingError, attempts: :unlimited
+
+      around_perform :continue
+    end
+
+    def step(step_name, start: nil, &block)
+      continuation.step(step_name, start: start) do |step|
+        if block_given?
+          block.call(step)
+        else
+          step_method = method(step_name)
+
+          raise ArgumentError, "Step method '#{step_name}' must accept 0 or 1 arguments" if step_method.arity > 1
+
+          if step_method.parameters.any? { |type, name| type == :key || type == :keyreq }
+            raise ArgumentError, "Step method '#{step_name}' must not accept keyword arguments"
+          end
+
+          step_method.arity == 0 ? step_method.call : step_method.call(step)
+        end
+      end
+    end
+
+    def serialize
+      super.merge(CONTINUATION_KEY => continuation.to_h)
+    end
+
+    def deserialize(job_data)
+      super
+      @continuation = Continuation.new(self, job_data.fetch(CONTINUATION_KEY, {}))
+    end
+
+    private
+      def continuation
+        @continuation ||= Continuation.new(self, {})
+      end
+
+      def continue(&block)
+        continuation.continue(&block)
+      end
+  end
+end