Support Active Job Continuations

.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