Update fork to upstream solid_queue v1.4.0

.github/workflows/main.yml

@@ -13,13 +13,15 @@ jobs:
         with:
           ruby-version: 3.4
           bundler-cache: true
+          cache-version: 1
       - name: Run rubocop
         run: |
           bundle exec rubocop --parallel
 
   tests:
     name: Tests
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     strategy:
       fail-fast: false
       matrix:
@@ -28,12 +30,17 @@ jobs:
         - 3.2
         - 3.3
         - 3.4
+        - 4.0
         database: [ mysql, postgres, sqlite ]
-        gemfile: [ rails_7_1, rails_7_2, rails_8_0, rails_main ]
+        gemfile: [ rails_7_1, rails_7_2, rails_8_0, rails_8_1, rails_main ]
         exclude:
           - ruby-version: "3.1"
             gemfile: rails_8_0
           - ruby-version: "3.1"
+            gemfile: rails_8_1
+          - ruby-version: "3.1"
+            gemfile: rails_main
+          - ruby-version: "3.2"
             gemfile: rails_main
     services:
       mysql:
@@ -61,6 +68,7 @@ jobs:
         with:
           ruby-version: ${{ matrix.ruby-version }}
           bundler-cache: true
+          cache-version: 1
       - name: Update to latest Rails
         run: |
           bundle update railties
@@ -77,4 +85,4 @@ jobs:
           path: |
             test/dummy/log/test.log
           if-no-files-found: ignore
-          retention-days: 7
+          retention-days: 30

Appraisals

@@ -15,6 +15,10 @@ appraise "rails-8-0" do
   gem "railties", "~> 8.0.0"
 end
 
+appraise "rails-8-1" do
+  gem "railties", "~> 8.1.0"
+end
+
 appraise "rails-main" do
   gem "railties", github: "rails/rails", branch: "main"
 end

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    solid_queue (1.2.2)
+    solid_queue (1.4.0)
       activejob (>= 7.1)
       activerecord (>= 7.1)
       concurrent-ruby (>= 1.3.1)
@@ -201,6 +201,7 @@ DEPENDENCIES
   appraisal
   debug (~> 1.9)
   logger
+  minitest (~> 5.0)
   mocha
   mysql2
   pg

Rakefile

@@ -5,7 +5,9 @@ require "bundler/setup"
 APP_RAKEFILE = File.expand_path("test/dummy/Rakefile", __dir__)
 load "rails/tasks/engine.rake"
 
-load "rails/tasks/statistics.rake"
+if Rails::VERSION::MAJOR < 8
+  load "rails/tasks/statistics.rake"
+end
 
 require "bundler/gem_tasks"
 require "rake/tasklib"

app/models/solid_queue/blocked_execution.rb

@@ -26,7 +26,9 @@ def release_many(concurrency_keys)
 
       def release_one(concurrency_key)
         transaction do
-          if execution = ordered.where(concurrency_key: concurrency_key).limit(1).non_blocking_lock.first
+          if execution = ordered.where(concurrency_key: concurrency_key).limit(1)
+              .use_index(:index_solid_queue_blocked_executions_for_release)
+              .non_blocking_lock.first
             execution.release
           end
         end

app/models/solid_queue/failed_execution.rb

@@ -6,7 +6,7 @@ class FailedExecution < Execution
 
     serialize :error, coder: JSON
 
-    before_create :expand_error_details_from_exception
+    before_save :expand_error_details_from_exception, if: :exception
 
     attr_accessor :exception
 

app/models/solid_queue/job/concurrency_controls.rb

@@ -26,7 +26,7 @@ def unblock_next_blocked_job
       end
 
       def concurrency_limited?
-        concurrency_key.present?
+        concurrency_key.present? && job_class.present?
       end
 
       def blocked?

app/models/solid_queue/job/retryable.rb

@@ -16,7 +16,16 @@ def retry
       end
 
       def failed_with(exception)
-        FailedExecution.create_or_find_by!(job_id: id, exception: exception)
+        FailedExecution.transaction(requires_new: true) do
+          FailedExecution.create!(job_id: id, exception: exception)
+        end
+      rescue ActiveRecord::RecordNotUnique
+        if (failed_execution = FailedExecution.find_by(job_id: id))
+          failed_execution.exception = exception
+          failed_execution.save!
+        else
+          retry
+        end
       end
 
       def reset_execution_counters

app/models/solid_queue/ready_execution.rb

@@ -30,7 +30,8 @@ def select_and_lock(queue_relation, process_id, limit)
         end
 
         def select_candidates(queue_relation, limit)
-          queue_relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id)
+          # Force query execution here with #to_a to avoid unintended FOR UPDATE query executions
+          queue_relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id).to_a
         end
 
         def lock_candidates(executions, process_id)

app/models/solid_queue/record.rb

@@ -20,6 +20,13 @@ def supports_insert_conflict_target?
           connection.supports_insert_conflict_target?
         end
       end
+
+      # Pass index hints to the query optimizer using SQL comment hints.
+      # Uses MySQL 8 optimizer hint query comments, which SQLite and
+      # PostgreSQL ignore.
+      def use_index(*indexes)
+        optimizer_hints "INDEX(#{quoted_table_name} #{indexes.join(', ')})"
+      end
     end
   end
 end

app/models/solid_queue/recurring_task.rb

@@ -6,11 +6,12 @@ module SolidQueue
   class RecurringTask < Record
     serialize :arguments, coder: Arguments, default: []
 
-    validate :supported_schedule
+    validate :ensure_schedule_supported
     validate :ensure_command_or_class_present
-    validate :existing_job_class
+    validate :ensure_existing_job_class
 
     scope :static, -> { where(static: true) }
+    scope :dynamic, -> { where(static: false) }
 
     has_many :recurring_executions, foreign_key: :task_key, primary_key: :key
 
@@ -32,7 +33,15 @@ def from_configuration(key, **options)
           queue_name: options[:queue].presence,
           priority: options[:priority].presence,
           description: options[:description],
-          static: true
+          static: options.fetch(:static, true)
+      end
+
+      def create_dynamic_task(key, **options)
+        from_configuration(key, **options.merge(static: false)).save!
+      end
+
+      def delete_dynamic_task(key)
+        RecurringTask.dynamic.find_by!(key: key).destroy
       end
 
       def create_or_update_all(tasks)
@@ -102,19 +111,28 @@ def attributes_for_upsert
     end
 
     private
-      def supported_schedule
+      def ensure_schedule_supported
         unless parsed_schedule.instance_of?(Fugit::Cron)
           errors.add :schedule, :unsupported, message: "is not a supported recurring schedule"
         end
+      rescue ArgumentError => error
+        message = if error.message.include?("multiple crons")
+          "generates multiple cron schedules. Please use separate recurring tasks for each schedule, " +
+          "or use explicit cron syntax (e.g., '40 0,15 * * *' for multiple times with the same minutes)"
+        else
+          error.message
+        end
+
+        errors.add :schedule, :unsupported, message: message
       end
 
       def ensure_command_or_class_present
         unless command.present? || class_name.present?
-          errors.add :base, :command_and_class_blank, message: "either command or class_name must be present"
+          errors.add :base, :command_and_class_blank, message: "either command or class must be present"
         end
       end
 
-      def existing_job_class
+      def ensure_existing_job_class
         if class_name.present? && job_class.nil?
           errors.add :class_name, :undefined, message: "doesn't correspond to an existing class"
         end
@@ -152,7 +170,7 @@ def arguments_with_kwargs
 
 
       def parsed_schedule
-        @parsed_schedule ||= Fugit.parse(schedule)
+        @parsed_schedule ||= Fugit.parse(schedule, multi: :fail)
       end
 
       def job_class

app/models/solid_queue/semaphore.rb

@@ -40,7 +40,7 @@ def initialize(job)
       end
 
       def wait
-        if semaphore = Semaphore.find_by(key: key)
+        if semaphore = Semaphore.lock.find_by(key: key)
           semaphore.value > 0 && attempt_decrement
         else
           attempt_creation

gemfiles/rails_8_1.gemfile

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

lib/puma/plugin/solid_queue.rb

@@ -1,44 +1,92 @@
 require "puma/plugin"
 
+module Puma
+  class DSL
+    def solid_queue_mode(mode = :fork)
+      @options[:solid_queue_mode] = mode.to_sym
+    end
+  end
+end
+
 Puma::Plugin.create do
   attr_reader :puma_pid, :solid_queue_pid, :log_writer, :solid_queue_supervisor
 
   def start(launcher)
     @log_writer = launcher.log_writer
     @puma_pid = $$
 
-    in_background do
-      monitor_solid_queue
+    if launcher.options[:solid_queue_mode] == :async
+      start_async(launcher)
+    else
+      start_forked(launcher)
     end
+  end
 
-    if Gem::Version.new(Puma::Const::VERSION) < Gem::Version.new("7")
-      launcher.events.on_booted do
-        @solid_queue_pid = fork do
-          Thread.new { monitor_puma }
-          SolidQueue::Supervisor.start
+  private
+    def start_forked(launcher)
+      in_background do
+        monitor_solid_queue
+      end
+
+      if Gem::Version.new(Puma::Const::VERSION) < Gem::Version.new("7")
+        launcher.events.on_booted do
+          @solid_queue_pid = fork do
+            Thread.new { monitor_puma }
+            SolidQueue::Supervisor.start(mode: :fork)
+          end
+        end
+
+        launcher.events.on_stopped { stop_solid_queue_fork }
+        launcher.events.on_restart { stop_solid_queue_fork }
+      else
+        launcher.events.after_booted do
+          @solid_queue_pid = fork do
+            Thread.new { monitor_puma }
+            start_solid_queue(mode: :fork)
+          end
         end
+
+        launcher.events.after_stopped { stop_solid_queue_fork }
+        launcher.events.before_restart { stop_solid_queue_fork }
       end
+    end
 
-      launcher.events.on_stopped { stop_solid_queue }
-      launcher.events.on_restart { stop_solid_queue }
-    else
-      launcher.events.after_booted do
-        @solid_queue_pid = fork do
-          Thread.new { monitor_puma }
-          SolidQueue::Supervisor.start
+    def start_async(launcher)
+      if Gem::Version.new(Puma::Const::VERSION) < Gem::Version.new("7")
+        launcher.events.on_booted do
+          start_solid_queue(mode: :async, standalone: false)
+        end
+
+        launcher.events.on_stopped { solid_queue_supervisor&.stop }
+
+        launcher.events.on_restart do
+          solid_queue_supervisor&.stop
+          start_solid_queue(mode: :async, standalone: false)
+        end
+      else
+        launcher.events.after_booted do
+          start_solid_queue(mode: :async, standalone: false)
+        end
+
+        launcher.events.after_stopped { solid_queue_supervisor&.stop }
+
+        launcher.events.before_restart do
+          solid_queue_supervisor&.stop
+          start_solid_queue(mode: :async, standalone: false)
         end
       end
+    end
 
-      launcher.events.after_stopped { stop_solid_queue }
-      launcher.events.before_restart { stop_solid_queue }
+    def start_solid_queue(**options)
+      @solid_queue_supervisor = SolidQueue::Supervisor.start(**options)
     end
-  end
 
-  private
-    def stop_solid_queue
+    def stop_solid_queue_fork
+      return unless solid_queue_pid
+
       Process.waitpid(solid_queue_pid, Process::WNOHANG)
       log "Stopping Solid Queue..."
-      Process.kill(:INT, solid_queue_pid) if solid_queue_pid
+      Process.kill(:INT, solid_queue_pid)
       Process.wait(solid_queue_pid)
     rescue Errno::ECHILD, Errno::ESRCH
     end
@@ -48,7 +96,7 @@ def monitor_puma
     end
 
     def monitor_solid_queue
-      monitor(:solid_queue_dead?, "Detected Solid Queue has gone away, stopping Puma...")
+      monitor(:solid_queue_fork_dead?, "Detected Solid Queue has gone away, stopping Puma...")
     end
 
     def monitor(process_dead, message)
@@ -62,7 +110,7 @@ def monitor(process_dead, message)
       end
     end
 
-    def solid_queue_dead?
+    def solid_queue_fork_dead?
       if solid_queue_started?
         Process.waitpid(solid_queue_pid, Process::WNOHANG)
       end

lib/solid_queue.rb

@@ -43,6 +43,14 @@ module SolidQueue
 
   delegate :on_start, :on_stop, :on_exit, to: Supervisor
 
+  def schedule_recurring_task(key, **options)
+    RecurringTask.create_dynamic_task(key, **options)
+  end
+
+  def unschedule_recurring_task(key)
+    RecurringTask.delete_dynamic_task(key)
+  end
+
   [ Dispatcher, Scheduler, Worker ].each do |process|
     define_singleton_method(:"on_#{process.name.demodulize.downcase}_start") do |&block|
       process.on_start(&block)