Merge pull request #277 from rails/fail-jobs-when-worker-is-killed

Fail in-progress jobs when the worker running them exits abnormally

Author: rosa

UPGRADING.md

@@ -1,3 +1,32 @@
+# Upgrading to version 0.6.x
+
+## New migration in 3 steps
+This version adds two new migrations to modify the `solid_queue_processes` table. The goal of that migration is to add a new column that needs to be `NOT NULL`. This needs to be done with two migrations and the following steps to ensure it happens without downtime and with new processes being able to register just fine:
+1. Run the first migration that adds the new column, nullable
+2. Deploy the updated Solid Queue code that uses this column
+2. Run the second migration. This migration does two things:
+  - Backfill existing rows that would have the column as NULL
+  - Make the column not nullable and add a new index
+
+Besides, it adds another migration with no effects to the `solid_queue_recurring_tasks` table. This one can be run just fine whenever, as the column affected is not used.
+
+To install the migrations:
+```bash
+$ bin/rails solid_queue:install:migrations
+```
+
+Or, if you're using a different database for Solid Queue:
+
+```bash
+$ bin/rails solid_queue:install:migrations DATABASE=<the_name_of_your_solid_queue_db>
+```
+
+And then follow the steps above, running first one, then deploying the code, then running the second one.
+
+## New behaviour when workers are killed
+From this version onwards, when a worker is killed and the supervisor can detect that, it'll fail in-progress jobs claimed by that worker. For this to work correctly, you need to run the above migration and ensure you restart any supervisors you'd have. 
+
+
 # Upgrading to version 0.5.x
 This version includes a new migration to improve recurring tasks. To install it, just run:
 

app/models/solid_queue/claimed_execution.rb

@@ -35,6 +35,18 @@ def release_all
       end
     end
 
+    def fail_all_with(error)
+      SolidQueue.instrument(:fail_many_claimed) do |payload|
+        includes(:job).tap do |executions|
+          payload[:size] = executions.size
+          payload[:process_ids] = executions.map(&:process_id).uniq
+          payload[:job_ids] = executions.map(&:job_id).uniq
+
+          executions.each { |execution| execution.failed_with(error) }
+        end
+      end
+    end
+
     def discard_all_in_batches(*)
       raise UndiscardableError, "Can't discard jobs in progress"
     end
@@ -69,6 +81,13 @@ def discard
     raise UndiscardableError, "Can't discard a job in progress"
   end
 
+  def failed_with(error)
+    transaction do
+      job.failed_with(error)
+      destroy!
+    end
+  end
+
   private
     def execute
       ActiveJob::Base.execute(job.arguments)
@@ -83,11 +102,4 @@ def finished
         destroy!
       end
     end
-
-    def failed_with(error)
-      transaction do
-        job.failed_with(error)
-        destroy!
-      end
-    end
 end

app/models/solid_queue/process.rb

@@ -13,10 +13,10 @@ def self.register(**attributes)
       create!(attributes.merge(last_heartbeat_at: Time.current)).tap do |process|
         payload[:process_id] = process.id
       end
+    rescue Exception => error
+      payload[:error] = error
+      raise
     end
-  rescue Exception => error
-    SolidQueue.instrument :register_process, **attributes.merge(error: error)
-    raise
   end
 
   def heartbeat
@@ -25,8 +25,6 @@ def heartbeat
 
   def deregister(pruned: false)
     SolidQueue.instrument :deregister_process, process: self, pruned: pruned do |payload|
-      payload[:claimed_size] = claimed_executions.size if claims_executions?
-
       destroy!
     rescue Exception => error
       payload[:error] = error

app/models/solid_queue/process/executor.rb

@@ -11,6 +11,12 @@ module Executor
         after_destroy -> { claimed_executions.release_all }, if: :claims_executions?
       end
 
+      def fail_all_claimed_executions_with(error)
+        if claims_executions?
+          claimed_executions.fail_all_with(error)
+        end
+      end
+
       private
         def claims_executions?
           kind == "Worker"

app/models/solid_queue/process/prunable.rb

@@ -1,6 +1,12 @@
 # frozen_string_literal: true
 
 module SolidQueue
+  class ProcessPrunedError < RuntimeError
+    def initialize(last_heartbeat_at)
+      super("Process was found dead and pruned (last heartbeat at: #{last_heartbeat_at}")
+    end
+  end
+
   class Process
     module Prunable
       extend ActiveSupport::Concern
@@ -15,11 +21,18 @@ def prune
             prunable.non_blocking_lock.find_in_batches(batch_size: 50) do |batch|
               payload[:size] += batch.size
 
-              batch.each { |process| process.deregister(pruned: true) }
+              batch.each(&:prune)
             end
           end
         end
       end
+
+      def prune
+        error = ProcessPrunedError.new(last_heartbeat_at)
+        fail_all_claimed_executions_with(error)
+
+        deregister(pruned: true)
+      end
     end
   end
 end

db/migrate/20240811173327_add_name_to_processes.rb

@@ -0,0 +1,5 @@
+class AddNameToProcesses < ActiveRecord::Migration[7.1]
+  def change
+    add_column :solid_queue_processes, :name, :string
+  end
+end

db/migrate/20240813160053_make_name_not_null.rb

@@ -0,0 +1,16 @@
+class MakeNameNotNull < ActiveRecord::Migration[7.1]
+  def up
+    SolidQueue::Process.where(name: nil).find_each do |process|
+      process.name ||= [ process.kind.downcase, SecureRandom.hex(10) ].join("-")
+      process.save!
+    end
+
+    change_column :solid_queue_processes, :name, :string, null: false
+    add_index :solid_queue_processes, [ :name, :supervisor_id ], unique: true
+  end
+
+  def down
+    remove_index :solid_queue_processes, [ :name, :supervisor_id ]
+    change_column :solid_queue_processes, :name, :string, null: false
+  end
+end

lib/solid_queue/configuration.rb

@@ -2,6 +2,12 @@
 
 module SolidQueue
   class Configuration
+    class Process < Struct.new(:kind, :attributes)
+      def instantiate
+        "SolidQueue::#{kind.to_s.titleize}".safe_constantize.new(**attributes)
+      end
+    end
+
     WORKER_DEFAULTS = {
       queues: "*",
       threads: 3,
@@ -22,28 +28,10 @@ def initialize(mode: :fork, load_from: nil)
       @raw_config = config_from(load_from)
     end
 
-    def processes
+    def configured_processes
       dispatchers + workers
     end
 
-    def workers
-      workers_options.flat_map do |worker_options|
-        processes = if mode.fork?
-          worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
-        else
-          WORKER_DEFAULTS[:processes]
-        end
-        processes.times.map { Worker.new(**worker_options.with_defaults(WORKER_DEFAULTS)) }
-      end
-    end
-
-    def dispatchers
-      dispatchers_options.map do |dispatcher_options|
-        recurring_tasks = parse_recurring_tasks dispatcher_options[:recurring_tasks]
-        Dispatcher.new **dispatcher_options.merge(recurring_tasks: recurring_tasks).with_defaults(DISPATCHER_DEFAULTS)
-      end
-    end
-
     def max_number_of_threads
       # At most "threads" in each worker + 1 thread for the worker + 1 thread for the heartbeat task
       workers_options.map { |options| options[:threads] }.max + 2
@@ -54,6 +42,24 @@ def max_number_of_threads
 
       DEFAULT_CONFIG_FILE_PATH = "config/solid_queue.yml"
 
+      def workers
+        workers_options.flat_map do |worker_options|
+          processes = if mode.fork?
+            worker_options.fetch(:processes, WORKER_DEFAULTS[:processes])
+          else
+            WORKER_DEFAULTS[:processes]
+          end
+          processes.times.map { Process.new(:worker, worker_options.with_defaults(WORKER_DEFAULTS)) }
+        end
+      end
+
+      def dispatchers
+        dispatchers_options.map do |dispatcher_options|
+          recurring_tasks = parse_recurring_tasks dispatcher_options[:recurring_tasks]
+          Process.new :dispatcher, dispatcher_options.merge(recurring_tasks: recurring_tasks).with_defaults(DISPATCHER_DEFAULTS)
+        end
+      end
+
       def config_from(file_or_hash, env: Rails.env)
         config = load_config_from(file_or_hash)
         config[env.to_sym] ? config[env.to_sym] : config

lib/solid_queue/log_subscriber.rb

@@ -12,11 +12,15 @@ def claim(event)
   end
 
   def release_many_claimed(event)
-    debug formatted_event(event, action: "Release claimed jobs", **event.payload.slice(:size))
+    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))
   end
 
   def release_claimed(event)
-    debug 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))
   end
 
   def retry_all(event)
@@ -63,7 +67,8 @@ def start_process(event)
     attributes = {
       pid: process.pid,
       hostname: process.hostname,
-      process_id: process.process_id
+      process_id: process.process_id,
+      name: process.name
     }.merge(process.metadata)
 
     info formatted_event(event, action: "Started #{process.kind}", **attributes)
@@ -75,15 +80,16 @@ def shutdown_process(event)
     attributes = {
       pid: process.pid,
       hostname: process.hostname,
-      process_id: process.process_id
+      process_id: process.process_id,
+      name: process.name
     }.merge(process.metadata)
 
     info formatted_event(event, action: "Shutdown #{process.kind}", **attributes)
   end
 
   def register_process(event)
     process_kind = event.payload[:kind]
-    attributes = event.payload.slice(:pid, :hostname, :process_id)
+    attributes = event.payload.slice(:pid, :hostname, :process_id, :name)
 
     if error = event.payload[:error]
       warn formatted_event(event, action: "Error registering #{process_kind}", **attributes.merge(error: formatted_error(error)))
@@ -99,6 +105,7 @@ def deregister_process(event)
       process_id: process.id,
       pid: process.pid,
       hostname: process.hostname,
+      name: process.name,
       last_heartbeat_at: process.last_heartbeat_at.iso8601,
       claimed_size: event.payload[:claimed_size],
       pruned: event.payload[:pruned]
@@ -147,7 +154,7 @@ def replace_fork(event)
       termsig: status.termsig
 
     if replaced_fork = event.payload[:fork]
-      info formatted_event(event, action: "Replaced terminated #{replaced_fork.kind}", **attributes.merge(hostname: replaced_fork.hostname))
+      info formatted_event(event, action: "Replaced terminated #{replaced_fork.kind}", **attributes.merge(hostname: replaced_fork.hostname, name: replaced_fork.name))
     else
       warn formatted_event(event, action: "Tried to replace forked process but it had already died", **attributes)
     end

lib/solid_queue/processes/base.rb

@@ -6,6 +6,12 @@ class Base
       include Callbacks # Defines callbacks needed by other concerns
       include AppExecutor, Registrable, Interruptible, Procline
 
+      attr_reader :name
+
+      def initialize(*)
+        @name = generate_name
+      end
+
       def kind
         self.class.name.demodulize
       end
@@ -21,6 +27,11 @@ def pid
       def metadata
         {}
       end
+
+      private
+        def generate_name
+          [ kind.downcase, SecureRandom.hex(10) ].join("-")
+        end
     end
   end
 end