Acquire concurrency locks by key when dispatching several jobs

Appraisals

@@ -2,6 +2,8 @@
 
 appraise "rails-7-1" do
   gem "railties", "~> 7.1.0"
+  # json 3 rejects the quirks_mode option this Active Support passes.
+  gem "json", "< 3"
 end
 
 appraise "rails-7-2" do
@@ -10,6 +12,8 @@ end
 
 appraise "rails-8-0" do
   gem "railties", "~> 8.0.0"
+  # json 3 rejects the quirks_mode option this Active Support passes.
+  gem "json", "< 3"
 end
 
 appraise "rails-8-1" do

README.md

@@ -401,6 +401,8 @@ The supervisor is in charge of managing these processes, and it responds to the
 
 When receiving a `QUIT` signal, if workers still have jobs in-flight, these will be returned to the queue when the processes are deregistered.
 
+In `async` mode, workers, dispatchers and the scheduler are threads inside the supervisor's process, so they can't be signaled one by one. On `TERM` or `INT`, the supervisor stops them and waits up to `SolidQueue.shutdown_timeout`. If any are still running by then, it deregisters itself, which returns their in-flight jobs to the queue, and exits. On `QUIT`, it exits immediately without deregistering anything, so in-flight jobs remain claimed until another supervisor prunes these processes and marks the jobs as failed with a `SolidQueue::Processes::ProcessPrunedError`.
+
 On Windows, the `QUIT` signal can't be trapped, so the supervisor only responds to `TERM` and `INT` there.
 
 If processes have no chance of cleaning up before exiting (e.g. if someone pulls a cable somewhere), in-flight jobs might remain claimed by the processes executing them. Processes send heartbeats, and the supervisor checks and prunes processes with expired heartbeats. Jobs that were claimed by processes with an expired heartbeat will be marked as failed with a `SolidQueue::Processes::ProcessPrunedError`. You can configure both the frequency of heartbeats and the threshold to consider a process dead. See the section below for this.
@@ -927,6 +929,8 @@ It's possible to run multiple schedulers with the same `recurring_tasks` configu
 
 **Note**: a single recurring schedule is supported, so you can have multiple schedulers using the same schedule, but not multiple schedulers using different configurations.
 
+When a scheduler starts, it removes any static recurring tasks that are no longer in its configuration. If you remove all of them, though, no scheduler starts, so the old tasks stay in the `solid_queue_recurring_tasks` table, and in Mission Control, even though nothing runs them. You can delete them from a console with `SolidQueue::RecurringTask.static.delete_all`.
+
 Finally, it's possible to configure jobs that aren't handled by Solid Queue. That is, you can have a job like this in your app:
 ```ruby
 class MyResqueJob < ApplicationJob

app/models/solid_queue/claimed_execution.rb

@@ -66,11 +66,11 @@ def fail_all_with(error)
       end
 
       def discard_all_in_batches(*)
-        raise UndiscardableError, "Can't discard jobs in progress"
+        raise SolidQueue::Execution::UndiscardableError, "Can't discard jobs in progress"
       end
 
       def discard_all_from_jobs(*)
-        raise UndiscardableError, "Can't discard jobs in progress"
+        raise SolidQueue::Execution::UndiscardableError, "Can't discard jobs in progress"
       end
     end
 

app/models/solid_queue/execution.rb

@@ -28,13 +28,15 @@ def execution_data_from_jobs(jobs)
       end
 
       def discard_all_in_batches(batch_size: 500)
-        pending = count
+        # Newly unblocked executions must survive later discard batches.
+        last_id = maximum(:id) || 0
+        pending = where(id: ..last_id).count
         discarded = 0
 
         SolidQueue.instrument(:discard_all, batch_size: batch_size, status: type, batches: 0, size: 0) do |payload|
           loop do
             transaction do
-              job_ids = limit(batch_size).order(:job_id).lock.pluck(:job_id)
+              job_ids = where(id: ..last_id).limit(batch_size).order(:job_id).lock.pluck(:job_id)
               discarded = discard_jobs job_ids
 
               where(job_id: job_ids).delete_all

app/models/solid_queue/failed_execution.rb

@@ -13,7 +13,10 @@ class FailedExecution < Execution
     def self.retry_all(jobs)
       SolidQueue.instrument(:retry_all, jobs_size: jobs.size) do |payload|
         transaction do
-          payload[:size] = dispatch_jobs lock_all_from_jobs(jobs)
+          job_ids = lock_all_from_jobs(jobs)
+          Job.where(id: job_ids).find_each(&:reset_execution_counters)
+
+          payload[:size] = dispatch_jobs(job_ids)
         end
       end
     end
@@ -33,11 +36,11 @@ def retry
     end
 
     private
-      JSON_OVERHEAD = 256
+      TRUNCATED_MESSAGE_OMISSION = "... (truncated)"
 
       def expand_error_details_from_exception
         if exception
-          self.error = { exception_class: exception_class_name, message: exception_message, backtrace: exception_backtrace }
+          self.error = fit_into_error_column(exception_class: exception_class_name, message: exception_message, backtrace: exception_backtrace)
         end
       end
 
@@ -50,32 +53,58 @@ def exception_message
       end
 
       def exception_backtrace
-        if (limit = determine_backtrace_size_limit) && exception.backtrace.to_json.bytesize > limit
-          truncate_backtrace(exception.backtrace, limit)
-        else
-          exception.backtrace
+        exception.backtrace
+      end
+
+      # Columns with a size limit, like MySQL's TEXT, reject (or truncate into invalid JSON)
+      # errors that don't fit. Drop backtrace lines first and, if that's not enough, cut the
+      # message, measuring the error as it'll be stored, after JSON escaping.
+      def fit_into_error_column(error)
+        if (limit = error_column_limit) && serialized_error_size(error) > limit
+          error[:backtrace] = truncate_backtrace(error, limit) if error[:backtrace]
+          error[:message] = truncate_message(error, limit) if serialized_error_size(error) > limit
         end
+
+        error
       end
 
-      def determine_backtrace_size_limit
+      def error_column_limit
         column = self.class.connection_pool.with_connection do |connection|
           connection.schema_cache.columns_hash(self.class.table_name)["error"]
         end
 
-        if column && column.limit.present?
-          column.limit - exception_class_name.bytesize - exception_message.bytesize - JSON_OVERHEAD
+        column&.limit
+      end
+
+      def serialized_error_size(error)
+        self.class.type_for_attribute(:error).serialize(error).bytesize
+      end
+
+      def truncate_backtrace(error, limit)
+        # Each kept line adds its own encoded size plus a separating comma
+        available = limit - serialized_error_size(error.merge(backtrace: [])) + 1
+
+        error[:backtrace].take_while do |line|
+          (available -= serialized_error_size(line) + 1) >= 0
         end
       end
 
-      def truncate_backtrace(lines, limit)
-        [].tap do |truncated_backtrace|
-          lines.each do |line|
-            if (truncated_backtrace << line).to_json.bytesize > limit
-              truncated_backtrace.pop
-              break
-            end
+      def truncate_message(error, limit)
+        message = error[:message].to_s
+
+        # Every byte cut from the message shrinks its encoded size by at least one byte,
+        # so cutting the overflow converges in very few iterations, even with escaping
+        while message.present? && (overflow = serialized_error_size(error.merge(message: message)) - limit) > 0
+          truncate_to = message.bytesize - overflow
+
+          message = if truncate_to > TRUNCATED_MESSAGE_OMISSION.bytesize
+            message.truncate_bytes(truncate_to, omission: TRUNCATED_MESSAGE_OMISSION)
+          else
+            ""
           end
         end
+
+        message
       end
   end
 end

app/models/solid_queue/job/concurrency_controls.rb

@@ -14,8 +14,8 @@ module ConcurrencyControls
       end
 
       class_methods do
-        def release_all_concurrency_locks(jobs)
-          Semaphore.signal_all(jobs.select(&:concurrency_limited?))
+        def unblock_next_blocked_jobs(jobs)
+          jobs.find_each(&:unblock_next_blocked_job)
         end
       end
 
@@ -63,7 +63,7 @@ def handle_concurrency_conflict
         end
 
         def block
-          BlockedExecution.create_or_find_by!(job_id: id)
+          BlockedExecution.create_or_find_by!(job: self)
         end
 
         def release_next_blocked_job

app/models/solid_queue/job/executable.rb

@@ -39,8 +39,12 @@ def dispatch_all_at_once(jobs)
             ReadyExecution.create_all_from_jobs jobs
           end
 
+          # Each job acquires its concurrency lock here, and holds it until the
+          # transaction dispatching the whole set ends. Going through them by
+          # concurrency key makes everyone dispatching at the same time acquire
+          # the locks they share in the same order, so they can't deadlock.
           def dispatch_all_one_by_one(jobs)
-            jobs.each(&:dispatch)
+            jobs.sort_by { |job| [ job.concurrency_key, job.id ] }.each(&:dispatch)
           end
 
           def successfully_dispatched(jobs)
@@ -106,7 +110,7 @@ def discard
 
       private
         def ready
-          ReadyExecution.create_or_find_by!(job_id: id)
+          ReadyExecution.create_or_find_by!(job: self)
         end
 
         def execution

app/models/solid_queue/job/schedulable.rb

@@ -38,7 +38,7 @@ def scheduled?
 
       private
         def schedule
-          ScheduledExecution.create_or_find_by!(job_id: id)
+          ScheduledExecution.create_or_find_by!(job: self)
         end
 
         def execution

app/models/solid_queue/queue_selector.rb

@@ -34,7 +34,7 @@ def queue_names
       def eligible_queues
         if include_all_queues? then all_queues
         else
-          in_raw_order(exact_names + prefixed_names)
+          in_raw_order(exact_names | prefixed_names)
         end
       end
 
@@ -53,12 +53,12 @@ def exact_names
       def prefixed_names
         if prefixes.empty? then []
         else
-          relation.where(([ "queue_name LIKE ?" ] * prefixes.count).join(" OR "), *prefixes).distinct_values_of(:queue_name)
+          relation.where(([ "queue_name LIKE ? ESCAPE '!'" ] * prefixes.count).join(" OR "), *prefixes).distinct_values_of(:queue_name)
         end
       end
 
       def prefixes
-        @prefixes ||= raw_queues.select { |queue| prefixed_name?(queue) }.map { |queue| queue.tr("*", "%") }
+        @prefixes ||= raw_queues.select { |queue| prefixed_name?(queue) }.map { |queue| relation.sanitize_sql_like(queue, "!").tr("*", "%") }
       end
 
       def exact_name?(queue)
@@ -74,9 +74,9 @@ def paused_queues
       end
 
       def in_raw_order(queues)
-        # Only need to sort if we have prefixes and more than one queue name.
-        # Exact names are selected in the same order as they're found
-        if queues.one? || prefixes.empty?
+        # Exact names are already in configuration order. Prefix matches always
+        # go through the literal filter so a single false LIKE hit is not returned.
+        if prefixes.empty?
           queues
         else
           queues = queues.dup
@@ -89,9 +89,7 @@ def delete_in_order(raw_queue, queues)
           queues.delete(raw_queue)
         elsif prefixed_name?(raw_queue)
           prefix = raw_queue.tr("*", "")
-          queues.select { |queue| queue.start_with?(prefix) }.tap do |matches|
-            queues -= matches
-          end
+          queues.extract! { |queue| queue.start_with?(prefix) }
         end
       end
   end

app/models/solid_queue/ready_execution.rb

@@ -45,7 +45,7 @@ def lock_candidates(executions, process_id)
 
 
         def discard_jobs(job_ids)
-          Job.release_all_concurrency_locks Job.where(id: job_ids)
+          Job.unblock_next_blocked_jobs Job.where(id: job_ids)
           super
         end
     end

app/models/solid_queue/semaphore.rb

@@ -14,10 +14,6 @@ def signal(job)
         Proxy.new(job).signal
       end
 
-      def signal_all(jobs)
-        Proxy.signal_all(jobs)
-      end
-
       # Requires a unique index on key
       def create_unique_by(attributes)
         if supports_insert_conflict_target?
@@ -31,16 +27,6 @@ def create_unique_by(attributes)
     end
 
     class Proxy
-      def self.signal_all(jobs)
-        # Guard against incrementing a semaphore's value beyond its limit. Jobs can
-        # have different limits, so group them and cap each group with `value < limit`.
-        jobs.group_by { |job| job.concurrency_limit || 1 }.each do |limit, grouped_jobs|
-          Semaphore.where(key: grouped_jobs.map(&:concurrency_key))
-            .where(value: ...limit)
-            .update_all("value = value + 1")
-        end
-      end
-
       def initialize(job)
         @job = job
       end

gemfiles/rails_7_1.gemfile

@@ -3,5 +3,6 @@
 source "https://rubygems.org"
 
 gem "railties", "~> 7.1.0"
+gem "json", "< 3"
 
 gemspec path: "../"

gemfiles/rails_8_0.gemfile

@@ -3,5 +3,6 @@
 source "https://rubygems.org"
 
 gem "railties", "~> 8.0.0"
+gem "json", "< 3"
 
 gemspec path: "../"

lib/solid_queue.rb

@@ -16,6 +16,8 @@
 loader.ignore("#{__dir__}/solid_queue/tasks.rb")
 loader.ignore("#{__dir__}/generators")
 loader.ignore("#{__dir__}/puma")
+# Only bin/jobs uses the CLI, and requires it. Eager loading it would load Thor in every process.
+loader.do_not_eager_load("#{__dir__}/solid_queue/cli.rb")
 loader.setup
 
 module SolidQueue

lib/solid_queue/async_supervisor.rb

@@ -38,12 +38,26 @@ def perform_graceful_termination
         process_instances.values.each(&:stop)
 
         Timer.wait_until(SolidQueue.shutdown_timeout, -> { all_processes_terminated? })
+
+        # The exit! that follows skips the shutdown callbacks, so release the jobs
+        # of the threads still running now instead of leaving them to be pruned
+        attempt_to_deregister unless all_processes_terminated?
       end
 
       def perform_immediate_termination
         exit!
       end
 
+      def attempt_to_deregister
+        wrap_in_app_executor do
+          # Not our own copy: it shares the threads' registrations in memory, and
+          # those of threads that already deregistered are destroyed and frozen
+          SolidQueue::Process.find_by(id: process_id)&.deregister
+        end
+      rescue StandardError => error
+        handle_thread_error(error)
+      end
+
       def all_processes_terminated?
         process_instances.values.none?(&:alive?)
       end