Support Active Job Continuations

[Active Job Continuations](rails/rails#55127)
will be added in Rails 8.1. Queue adapters need to implement the
`stopping?` method to support them.

Implement the method and set it to true when a worker is stopping.

Also add Rails versions so we can test against Rails main. To prevent an
explosion of combinations, I've switched to only testing against minor
versions of Ruby.

Author: djmb

.github/workflows/main.yml

@@ -24,18 +24,17 @@ jobs:
       fail-fast: false
       matrix:
         ruby-version:
-        - 3.1.6
-        - 3.2.0
-        - 3.2.4
-        - 3.3.0
-        - 3.3.1
-        - 3.3.2
-        - 3.3.4
-        - 3.3.5
-        - 3.3.6
-        - 3.4.0
-        - 3.4.1
+        - 3.1
+        - 3.2
+        - 3.3
+        - 3.4
         database: [ mysql, postgres, sqlite ]
+        gemfile: [ rails_7_1, rails_7_2, rails_8_0, rails_main ]
+        exclude:
+          - ruby-version: "3.1"
+            gemfile: rails_8_0
+          - ruby-version: "3.1"
+            gemfile: rails_main
     services:
       mysql:
         image: mysql:8.0.31
@@ -52,6 +51,7 @@ jobs:
           - 55432:5432
     env:
       TARGET_DB: ${{ matrix.database }}
+      BUNDLE_GEMFILE: ${{ github.workspace }}/gemfiles/${{ matrix.gemfile }}.gemfile
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
@@ -60,6 +60,9 @@ jobs:
         with:
           ruby-version: ${{ matrix.ruby-version }}
           bundler-cache: true
+      - name: Update to latest Rails
+        run: |
+          bundle update railties
       - name: Setup test database
         run: |
           bin/rails db:setup

.gitignore

@@ -1,5 +1,6 @@
 /.bundle/
 /doc/
+/gemfiles/*.lock
 /log/*.log
 /pkg/
 /tmp/

Appraisals

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+appraise "rails-7-1" do
+  # rdoc 6.14 is not compatible with Ruby 3.1
+  gem 'rdoc', '6.13'
+  gem "railties", "~> 7.1.0"
+end
+
+appraise "rails-7-2" do
+  gem 'rdoc', '6.13'
+  gem "railties", "~> 7.2.0"
+end
+
+appraise "rails-8-0" do
+  gem "railties", "~> 8.0.0"
+end
+
+appraise "rails-main" do
+  gem "railties", github: "rails/rails", branch: "main"
+end

Gemfile.lock

@@ -50,6 +50,10 @@ GEM
       mutex_m
       securerandom (>= 0.3)
       tzinfo (~> 2.0)
+    appraisal (2.5.0)
+      bundler
+      rake
+      thor (>= 0.14.0)
     ast (2.4.2)
     base64 (0.2.0)
     benchmark (0.4.0)
@@ -189,6 +193,7 @@ PLATFORMS
   x86_64-linux
 
 DEPENDENCIES
+  appraisal
   debug (~> 1.9)
   logger
   mocha

gemfiles/rails_7_1.gemfile

@@ -0,0 +1,8 @@
+# This file was generated by Appraisal
+
+source "https://rubygems.org"
+
+gem "rdoc", "6.13"
+gem "railties", "~> 7.1.0"
+
+gemspec path: "../"

gemfiles/rails_7_2.gemfile

@@ -0,0 +1,8 @@
+# This file was generated by Appraisal
+
+source "https://rubygems.org"
+
+gem "rdoc", "6.13"
+gem "railties", "~> 7.2.0"
+
+gemspec path: "../"

gemfiles/rails_8_0.gemfile

@@ -0,0 +1,7 @@
+# This file was generated by Appraisal
+
+source "https://rubygems.org"
+
+gem "railties", "~> 8.0.0"
+
+gemspec path: "../"

gemfiles/rails_main.gemfile

@@ -0,0 +1,7 @@
+# This file was generated by Appraisal
+
+source "https://rubygems.org"
+
+gem "railties", branch: "main", git: "https://github.com/rails/rails.git"
+
+gemspec path: "../"

lib/active_job/queue_adapters/solid_queue_adapter.rb

@@ -7,7 +7,10 @@ module QueueAdapters
     # To use it set the queue_adapter config to +:solid_queue+.
     #
     #   Rails.application.config.active_job.queue_adapter = :solid_queue
-    class SolidQueueAdapter
+    class SolidQueueAdapter < (Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1 ? Object : AbstractAdapter)
+      class_attribute :stopping, default: false, instance_writer: false
+      SolidQueue.on_worker_stop { self.stopping = true }
+
       def enqueue_after_transaction_commit?
         true
       end

solid_queue.gemspec

@@ -32,6 +32,7 @@ Gem::Specification.new do |spec|
   spec.add_dependency "fugit", "~> 1.11.0"
   spec.add_dependency "thor", "~> 1.3.1"
 
+  spec.add_development_dependency "appraisal"
   spec.add_development_dependency "debug", "~> 1.9"
   spec.add_development_dependency "mocha"
   spec.add_development_dependency "puma"

test/dummy/app/jobs/continuable_job.rb

@@ -0,0 +1,22 @@
+begin
+  require "active_job/continuation"
+rescue LoadError
+  # Zeitwerk requires that we define the constant
+  class ContinuableJob; end
+  return
+end
+
+class ContinuableJob < ApplicationJob
+  include ActiveJob::Continuable
+
+  def perform(result, pause: 0)
+    step :step_one do
+      sleep pause if pause > 0
+      result.update!(queue_name: queue_name, status: "stepped", value: "step_one")
+    end
+    step :step_two do
+      sleep pause if pause > 0
+      result.update!(queue_name: queue_name, status: "stepped", value: "step_two")
+    end
+  end
+end

test/integration/concurrency_controls_test.rb

@@ -180,6 +180,9 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   end
 
   test "verify transactions remain valid after Job creation conflicts via limits_concurrency" do
+    # Doesn't work with enqueue_after_transaction_commit? true on SolidQueueAdapter, but only Rails 7.2 uses this
+    skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 2
+
     ActiveRecord::Base.transaction do
       SequentialUpdateResultJob.perform_later(@result, name: "A", pause: 0.2.seconds)
       SequentialUpdateResultJob.perform_later(@result, name: "B")

test/integration/continuation_test.rb

@@ -0,0 +1,65 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+begin
+  require "active_job/continuation"
+rescue LoadError
+  return
+end
+
+class ContinuationTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  def setup
+    start_processes
+    @result = JobResult.create!
+  end
+
+  teardown do
+    terminate_process(@pid) if process_exists?(@pid)
+  end
+
+  test "continuable job completes" do
+    ContinuableJob.perform_later(@result)
+
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert_no_unfinished_jobs
+    assert_last_step :step_two
+  end
+
+  test "continuable job can be interrupted and resumed" do
+    job = ContinuableJob.perform_later(@result, pause: 0.5.seconds)
+
+    sleep 0.2.seconds
+    signal_process(@pid, :TERM)
+
+    wait_for_jobs_to_be_released_for(2.seconds)
+
+    assert_no_claimed_jobs
+    assert_unfinished_jobs job
+    assert_last_step :step_one
+
+    ActiveJob::QueueAdapters::SolidQueueAdapter.stopping = false
+    start_processes
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert_no_unfinished_jobs
+    assert_last_step :step_two
+  end
+
+  private
+    def assert_last_step(step)
+      @result.reload
+      assert_equal "stepped", @result.status
+      assert_equal step.to_s, @result.value
+    end
+
+    def start_processes
+      default_worker = { queues: "default", polling_interval: 0.1, processes: 3, threads: 2 }
+      dispatcher = { polling_interval: 0.1, batch_size: 200, concurrency_maintenance_interval: 1 }
+      @pid = run_supervisor_as_fork(workers: [ default_worker ], dispatchers: [ dispatcher ])
+      wait_for_registered_processes(5, timeout: 5.second) # 3 workers working the default queue + dispatcher + supervisor
+    end
+end