Link claimed_executions to processes via process_name

.github/workflows/main.yml

@@ -29,10 +29,12 @@ jobs:
         - 3.3
         - 3.4
         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
     services:
@@ -52,6 +54,7 @@ jobs:
     env:
       TARGET_DB: ${{ matrix.database }}
       BUNDLE_GEMFILE: ${{ github.workspace }}/gemfiles/${{ matrix.gemfile }}.gemfile
+      RAILS_ENV: test
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
@@ -68,3 +71,12 @@ jobs:
           bin/rails db:setup
       - name: Run tests
         run: bin/rails test
+      - name: Upload logs on failure
+        if: ${{ failure() }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: logs-${{ matrix.database }}-${{ matrix.gemfile }}-ruby${{ matrix.ruby-version }}-attempt${{ github.run_attempt }}
+          path: |
+            test/dummy/log/test.log
+          if-no-files-found: ignore
+          retention-days: 30

.rubocop.yml

@@ -7,3 +7,4 @@ AllCops:
   TargetRubyVersion: 3.3
   Exclude:
     - "**/*_schema.rb"
+    - "lib/generators/solid_queue/update/templates/db/*"

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.2.4)
       activejob (>= 7.1)
       activerecord (>= 7.1)
       concurrent-ruby (>= 1.3.1)

README.md

@@ -342,7 +342,7 @@ When receiving a `QUIT` signal, if workers still have jobs in-flight, these will
 
 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.
 
-In a similar way, if a worker is terminated in any other way not initiated by the above signals (e.g. a worker is sent a `KILL` signal), jobs in progress will be marked as failed so that they can be inspected, with a `SolidQueue::Processes::Process::ProcessExitError`. Sometimes a job in particular is responsible for this, for example, if it has a memory leak and you have a mechanism to kill processes over a certain memory threshold, so this will help identifying this kind of situation.
+In a similar way, if a worker is terminated in any other way not initiated by the above signals (e.g. a worker is sent a `KILL` signal), jobs in progress will be marked as failed so that they can be inspected, with a `SolidQueue::Processes::ProcessExitError`. Sometimes a job in particular is responsible for this, for example, if it has a memory leak and you have a mechanism to kill processes over a certain memory threshold, so this will help identifying this kind of situation.
 
 
 ### Database configuration

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"

UPGRADING.md

@@ -1,4 +1,21 @@
-# Upgrading to version 1.x
+# Upgrading to version 1.3
+There's a new migration in this version that can be installed via:
+```bash
+bin/rails solid_queue:update
+```
+which is a new generator to facilitate updates. This will use the `queue` database by default, but if you're using a different database name for Solid Queue, you can install the new migrations in the right place with:
+```bash
+DATABASE=your-solid-queue-db-name bin/rails solid_queue:update
+```
+
+Finally, the migration needs to be run with:
+```bash
+bin/rails db:migrate
+```
+
+The migration affects the tables `solid_queue_claimed_executions` and `solid_queue_processes` tables. It's not mandatory: everything will continue working as before without it, only a deprecation warning will be emitted. The migration will be mandatory in the next major version (2.0).
+
+# Upgrading to version >=1.0, < 1.3
 The value returned for `enqueue_after_transaction_commit?` has changed to `true`, and it's no longer configurable. If you want to change this, you need to use Active Job's configuration options.
 
 # Upgrading to version 0.9.x

app/models/solid_queue/claimed_execution.rb

@@ -1,7 +1,20 @@
 # frozen_string_literal: true
 
 class SolidQueue::ClaimedExecution < SolidQueue::Execution
-  belongs_to :process
+  def self.process_name_column_exists?
+    table_exists? && column_names.include?("process_name")
+  rescue ActiveRecord::Tenanted::TenantDoesNotExistError
+    true
+  end
+
+  if process_name_column_exists?
+    belongs_to :process, primary_key: :name, foreign_key: :process_name
+  else
+    warn_about_pending_migrations if table_exists?
+
+    belongs_to :process
+    attr_accessor :process_name
+  end
 
   scope :orphaned, -> { where.missing(:process) }
 
@@ -12,12 +25,16 @@ def success?
   end
 
   class << self
-    def claiming(job_ids, process_id, &block)
-      job_data = Array(job_ids).collect { |job_id| { job_id: job_id, process_id: process_id } }
+    def claiming(job_ids, process, &block)
+      process_data = { process_id: process.id }.tap do |hsh|
+        hsh[:process_name] = process.name if process_name_column_exists?
+      end
+
+      job_data = Array(job_ids).collect { |job_id| { job_id: job_id }.merge(process_data) }
 
-      SolidQueue.instrument(:claim, process_id: process_id, job_ids: job_ids) do |payload|
+      SolidQueue.instrument(:claim, job_ids: job_ids, **process_data) do |payload|
         insert_all!(job_data)
-        where(job_id: job_ids, process_id: process_id).load.tap do |claimed|
+        where(job_id: job_ids, process_id: process.id).load.tap do |claimed|
           block.call(claimed)
 
           payload[:size] = claimed.size
@@ -46,7 +63,8 @@ def fail_all_with(error)
             execution.unblock_next_job
           end
 
-          payload[:process_ids] = executions.map(&:process_id).uniq
+          payload[:process_ids] = executions.map(&:process_id).uniq.presence
+          payload[:process_names] = executions.map(&:process_name).uniq.presence
           payload[:job_ids] = executions.map(&:job_id).uniq
           payload[:size] = executions.size
         end
@@ -76,7 +94,7 @@ def perform
   end
 
   def release
-    SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id) do
+    SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id, process_name: process_name) do
       transaction do
         job.dispatch_bypassing_concurrency_limits
         destroy!

app/models/solid_queue/process/executor.rb

@@ -6,7 +6,13 @@ module Executor
       extend ActiveSupport::Concern
 
       included do
-        has_many :claimed_executions
+        if ClaimedExecution.process_name_column_exists?
+          has_many :claimed_executions, primary_key: :name, foreign_key: :process_name
+        else
+          warn_about_pending_migrations if ClaimedExecution.table_exists?
+
+          has_many :claimed_executions
+        end
 
         after_destroy :release_all_claimed_executions
       end

app/models/solid_queue/ready_execution.rb

@@ -7,9 +7,9 @@ class ReadyExecution < Execution
     assumes_attributes_from_job
 
     class << self
-      def claim(queue_list, limit, process_id)
+      def claim(queue_list, limit, process)
         QueueSelector.new(queue_list, self).scoped_relations.flat_map do |queue_relation|
-          select_and_lock(queue_relation, process_id, limit).tap do |locked|
+          select_and_lock(queue_relation, process, limit).tap do |locked|
             limit -= locked.size
           end
         end
@@ -20,23 +20,23 @@ def aggregated_count_across(queue_list)
       end
 
       private
-        def select_and_lock(queue_relation, process_id, limit)
+        def select_and_lock(queue_relation, process, limit)
           return [] if limit <= 0
 
           transaction do
             candidates = select_candidates(queue_relation, limit)
-            lock_candidates(candidates, process_id)
+            lock_candidates(candidates, process)
           end
         end
 
         def select_candidates(queue_relation, limit)
           queue_relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id)
         end
 
-        def lock_candidates(executions, process_id)
+        def lock_candidates(executions, process)
           return [] if executions.none?
 
-          SolidQueue::ClaimedExecution.claiming(executions.map(&:job_id), process_id) do |claimed|
+          SolidQueue::ClaimedExecution.claiming(executions.map(&:job_id), process) do |claimed|
             ids_to_delete = executions.index_by(&:job_id).values_at(*claimed.map(&:job_id)).map(&:id)
             where(id: ids_to_delete).delete_all
           end

app/models/solid_queue/record.rb

@@ -20,6 +20,18 @@ def supports_insert_conflict_target?
           connection.supports_insert_conflict_target?
         end
       end
+
+      def warn_about_pending_migrations
+        SolidQueue.deprecator.warn(<<~DEPRECATION)
+          Solid Queue has pending database migrations. To get the new migration files, run:
+            rails solid_queue:update
+          which will install the migration under `db/queue_migrate`. To change the database, run
+            DATABASE=your-solid-queue-db rails solid_queue:update
+          Then, apply the migrations with:
+            rails db:migrate
+          These migrations will be required after version 2.0
+        DEPRECATION
+      end
     end
   end
 end

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/generators/solid_queue/install/templates/db/queue_schema.rb

@@ -15,8 +15,10 @@
     t.bigint "job_id", null: false
     t.bigint "process_id"
     t.datetime "created_at", null: false
+    t.string "process_name"
     t.index [ "job_id" ], name: "index_solid_queue_claimed_executions_on_job_id", unique: true
     t.index [ "process_id", "job_id" ], name: "index_solid_queue_claimed_executions_on_process_id_and_job_id"
+    t.index [ "process_name" ], name: "index_solid_queue_claimed_executions_on_process_name"
   end
 
   create_table "solid_queue_failed_executions", force: :cascade do |t|
@@ -60,7 +62,7 @@
     t.datetime "created_at", null: false
     t.string "name", null: false
     t.index [ "last_heartbeat_at" ], name: "index_solid_queue_processes_on_last_heartbeat_at"
-    t.index [ "name", "supervisor_id" ], name: "index_solid_queue_processes_on_name_and_supervisor_id", unique: true
+    t.index [ "name" ], name: "index_solid_queue_processes_on_name", unique: true
     t.index [ "supervisor_id" ], name: "index_solid_queue_processes_on_supervisor_id"
   end
 

lib/generators/solid_queue/update/templates/db/link_claimed_executions_with_processes_through_name.rb

@@ -0,0 +1,30 @@
+class LinkClaimedExecutionsWithProcessesThroughName < ActiveRecord::Migration[<%= ActiveRecord::VERSION::STRING.to_f %>]
+  def up
+    unless connection.column_exists?(:solid_queue_claimed_executions, :process_name)
+      add_column :solid_queue_claimed_executions, :process_name, :string
+      add_index :solid_queue_claimed_executions, :process_name
+    end
+
+    unless connection.index_exists?(:solid_queue_processes, :name)
+      add_index :solid_queue_processes, :name, unique: true
+    end
+
+    if connection.index_exists?(:solid_queue_processes, [ :name, :supervisor_id ])
+      remove_index :solid_queue_processes, [ :name, :supervisor_id ]
+    end
+  end
+
+  def down
+    if connection.column_exists?(:solid_queue_claimed_executions, :process_name)
+      remove_column :solid_queue_claimed_executions, :process_name
+    end
+
+    if connection.index_exists?(:solid_queue_processes, :name)
+      remove_index :solid_queue_processes, :name
+    end
+
+    unless connection.index_exists?(:solid_queue_processes, [ :name, :supervisor_id ])
+      add_index :solid_queue_processes, [ :name, :supervisor_id ], unique: true
+    end
+  end
+end

lib/generators/solid_queue/update/update_generator.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+require "rails/generators/active_record"
+
+class SolidQueue::UpdateGenerator < Rails::Generators::Base
+  include ActiveRecord::Generators::Migration
+
+  source_root File.expand_path("templates", __dir__)
+
+  class_option :database, type: :string, aliases: %i[ --db ], default: "queue",
+    desc: "The database that Solid Queue uses. Defaults to `queue`"
+
+  def copy_new_migrations
+    template_dir = Dir.new(File.join(self.class.source_root, "db"))
+
+    template_dir.each_child do |migration_file|
+      migration_template File.join("db", migration_file), File.join(db_migrate_path, migration_file), skip: true
+    end
+  end
+end

lib/solid_queue.rb

@@ -69,6 +69,10 @@ def preserve_finished_jobs?
     preserve_finished_jobs
   end
 
+  def deprecator
+    @deprecator ||= ActiveSupport::Deprecation.new(next_major_version, "SolidQueue")
+  end
+
   def instrument(channel, **options, &block)
     ActiveSupport::Notifications.instrument("#{channel}.solid_queue", **options, &block)
   end

lib/solid_queue/configuration.rb

@@ -188,6 +188,7 @@ def load_config_from_file(file)
         if file.exist?
           ActiveSupport::ConfigurationFile.parse(file).deep_symbolize_keys
         else
+          puts "[solid_queue] WARNING: Provided configuration file '#{file}' does not exist. Falling back to default configuration."
           {}
         end
       end

lib/solid_queue/engine.rb

@@ -37,5 +37,9 @@ class Engine < ::Rails::Engine
         include ActiveJob::ConcurrencyControls
       end
     end
+
+    initializer "solid_queue.deprecator" do |app|
+      app.deprecators[:solid_queue] = SolidQueue.deprecator
+    end
   end
 end

lib/solid_queue/log_subscriber.rb

@@ -8,19 +8,19 @@ def dispatch_scheduled(event)
   end
 
   def claim(event)
-    debug formatted_event(event, action: "Claim jobs", **event.payload.slice(:process_id, :job_ids, :claimed_job_ids, :size))
+    debug formatted_event(event, action: "Claim jobs", **event.payload.slice(:process_id, :process_name, :job_ids, :claimed_job_ids, :size))
   end
 
   def release_many_claimed(event)
     info formatted_event(event, action: "Release claimed jobs", **event.payload.slice(:size))
   end
 
   def fail_many_claimed(event)
-    warn formatted_event(event, action: "Fail claimed jobs", **event.payload.slice(:job_ids, :process_ids))
+    warn formatted_event(event, action: "Fail claimed jobs", **event.payload.slice(:job_ids, :process_ids, :process_names))
   end
 
   def release_claimed(event)
-    info formatted_event(event, action: "Release claimed job", **event.payload.slice(:job_id, :process_id))
+    info formatted_event(event, action: "Release claimed job", **event.payload.slice(:job_id, :process_id, :process_name))
   end
 
   def retry_all(event)

lib/solid_queue/processes/registrable.rb

@@ -18,17 +18,19 @@ def process_id
       attr_accessor :process
 
       def register
-        @process = SolidQueue::Process.register \
-          kind: kind,
-          name: name,
-          pid: pid,
-          hostname: hostname,
-          supervisor: try(:supervisor),
-          metadata: metadata.compact
+        wrap_in_app_executor do
+          @process = SolidQueue::Process.register \
+            kind: kind,
+            name: name,
+            pid: pid,
+            hostname: hostname,
+            supervisor: try(:supervisor),
+            metadata: metadata.compact
+        end
       end
 
       def deregister
-        process&.deregister
+        wrap_in_app_executor { process&.deregister }
       end
 
       def registered?