Support Active Job Continuations

djmb · rosa · commit 9bd2a1bc149e · 2025-06-10T18:36:07.000+03:00
[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.
diff --git a/.github/workflows/main.yml b/.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
diff --git a/.gitignore b/.gitignore
@@ -1,5 +1,6 @@
 /.bundle/
 /doc/
+/gemfiles/*.lock
 /log/*.log
 /pkg/
 /tmp/
diff --git a/Appraisals b/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
diff --git a/Gemfile.lock b/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
diff --git a/gemfiles/rails_7_1.gemfile b/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: "../"
diff --git a/gemfiles/rails_7_2.gemfile b/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: "../"
diff --git a/gemfiles/rails_8_0.gemfile b/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: "../"
diff --git a/gemfiles/rails_main.gemfile b/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: "../"
diff --git a/lib/active_job/queue_adapters/solid_queue_adapter.rb b/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
diff --git a/solid_queue.gemspec b/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"
diff --git a/test/dummy/app/jobs/continuable_job.rb b/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
diff --git a/test/integration/concurrency_controls_test.rb b/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")
diff --git a/test/integration/continuation_test.rb 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
diff --git a/test/integration/lifecycle_hooks_test.rb b/test/integration/lifecycle_hooks_test.rb
diff --git a/test/models/solid_queue/job_test.rb b/test/models/solid_queue/job_test.rb
diff --git a/test/test_helper.rb b/test/test_helper.rb
diff --git a/test/test_helpers/jobs_test_helper.rb b/test/test_helpers/jobs_test_helper.rb
