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

This applies to:
- Killed workers that the supervisor detects as dead.
- Reaped workers without a clear exit status.
- Orphaned executions that somehow lost their worker.
- Workers whose heartbeat expired.

Author: rosa

UPGRADING.md

@@ -1,3 +1,28 @@
+# Upgrading to version 0.6.x
+
+## New migration in 3 steps
+This version adds a new migration to the `solid_queue_processes` table. This migration adds a new column that needs to be `NOT NULL`. It will run in three steps:
+1. Add the new column, nullable
+2. Backfill existing rows that would have the column as NULL
+3. Make the column not nullable and add a new index
+
+To install it:
+```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 just run it.
+
+## 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

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)

lib/solid_queue/supervisor/fork_supervisor.rb

@@ -1,6 +1,18 @@
 # frozen_string_literal: true
 
 module SolidQueue
+  class ProcessExitError < RuntimeError
+    def initialize(status)
+      message = case
+      when status.exitstatus.present? then "Process pid=#{status.pid} exited with status #{status.exitstatus}"
+      when status.signaled? then "Process pid=#{status.pid} received unhandled signal #{status.termsig}"
+      else "Process pid=#{status.pid} exited unexpectedly"
+      end
+
+      super(message)
+    end
+  end
+
   class Supervisor::ForkSupervisor < Supervisor
     include Signals, Pidfiled
 
@@ -91,7 +103,10 @@ def reap_terminated_forks
           pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG)
           break unless pid
 
-          forks.delete(pid)
+          if (terminated_fork = forks.delete(pid)) && !status.exited? || status.exitstatus > 0
+            handle_claimed_jobs_by(terminated_fork, status)
+          end
+
           configured_processes.delete(pid)
         end
       rescue SystemCallError
@@ -100,14 +115,22 @@ def reap_terminated_forks
 
       def replace_fork(pid, status)
         SolidQueue.instrument(:replace_fork, supervisor_pid: ::Process.pid, pid: pid, status: status) do |payload|
-          if supervised_fork = forks.delete(pid)
-            payload[:fork] = supervised_fork
+          if terminated_fork = forks.delete(pid)
+            payload[:fork] = terminated_fork
+            handle_claimed_jobs_by(terminated_fork, status)
 
             start_process(configured_processes.delete(pid))
           end
         end
       end
 
+      def handle_claimed_jobs_by(terminated_fork, status)
+        if registered_process = process.supervisees.find_by(name: terminated_fork.name)
+          error = ProcessExitError.new(status)
+          registered_process.fail_all_claimed_executions_with(error)
+        end
+      end
+
       def all_forks_terminated?
         forks.empty?
       end

lib/solid_queue/supervisor/maintenance.rb

@@ -1,9 +1,15 @@
 module SolidQueue
+  class ProcessMissingError < RuntimeError
+    def initialize
+      super("The process that was running this job no longer exists")
+    end
+  end
+
   module Supervisor::Maintenance
     extend ActiveSupport::Concern
 
     included do
-      after_boot :release_orphaned_executions
+      after_boot :fail_orphaned_executions
     end
 
     private
@@ -27,8 +33,10 @@ def prune_dead_processes
         wrap_in_app_executor { SolidQueue::Process.prune }
       end
 
-      def release_orphaned_executions
-        wrap_in_app_executor { SolidQueue::ClaimedExecution.orphaned.release_all }
+      def fail_orphaned_executions
+        wrap_in_app_executor do
+          SolidQueue::ClaimedExecution.orphaned.fail_all_with(ProcessMissingError.new)
+        end
       end
   end
 end

test/integration/forked_processes_lifecycle_test.rb

@@ -190,6 +190,11 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
     assert process_exists?(@pid)
     terminate_process(@pid)
 
+    # Since the worker exited abnormally, the jobs it had claimed would be failed now
+    [ exit_job, pause_job ].each do |job|
+      assert_job_status(job, :failed)
+    end
+
     assert_clean_termination
   end
 
@@ -240,10 +245,10 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
     # And there's a new worker that has been registered for the background queue
     wait_for_registered_processes(4, timeout: 5.second)
 
-    # The job in the background queue was left claimed as the worker couldn't
-    # finish orderly
+    # The job in the background queue would be failed by the supervisor
+    # when it replaced the killed worker
     assert_started_job_result("killed_pause")
-    assert_job_status(killed_pause, :claimed)
+    assert_job_status(killed_pause, :failed)
     # The other one could finish
     assert_completed_job_results("pause", :default)
 

test/integration/instrumentation_test.rb

@@ -89,7 +89,7 @@ class InstrumentationTest < ActiveSupport::TestCase
     assert_equal 2, events.size
     register_event, deregister_event = events
     assert_event register_event, "register_process", kind: "Worker", pid: ::Process.pid, process_id: process.id
-    assert_event deregister_event, "deregister_process", process: process, pruned: false, claimed_size: 1
+    assert_event deregister_event, "deregister_process", process: process, pruned: false
   end
 
   test "starting and stopping a dispatcher emits register_process and deregister_process events" do
@@ -112,11 +112,7 @@ class InstrumentationTest < ActiveSupport::TestCase
   end
 
   test "pruning processes emit prune_processes and deregister_process events" do
-    time = Time.now
-    processes = 3.times.collect { |i| SolidQueue::Process.create!(kind: "Worker", supervisor_id: 42, pid: 10 + i, hostname: "localhost", last_heartbeat_at: time, name: "worker-123#{i}") }
-
-    # Heartbeats will expire
-    travel_to 3.days.from_now
+    processes = 3.times.collect { |i| SolidQueue::Process.create!(kind: "Worker", supervisor_id: 42, pid: 10 + i, hostname: "localhost", last_heartbeat_at: 3.days.ago, name: "worker-123#{i}") }
 
     events = subscribed(/.*process.*\.solid_queue/) do
       SolidQueue::Process.prune
@@ -129,10 +125,26 @@ class InstrumentationTest < ActiveSupport::TestCase
 
     assert_event prune_event, "prune_processes", size: 3
     deregister_events.each_with_index do |event, i|
-      assert_event event, "deregister_process", process: processes[i], pruned: true, claimed_size: 0
+      assert_event event, "deregister_process", process: processes[i], pruned: true
     end
   end
 
+  test "pruning a process with claimed executions emits fail_many_claimed event" do
+    process = SolidQueue::Process.create!(kind: "Worker", supervisor_id: 42, pid: 10, last_heartbeat_at: 3.days.ago, name: "worker-123")
+
+    3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) }
+    jobs = SolidQueue::Job.last(3)
+
+    SolidQueue::ReadyExecution.claim("*", 5, process.id)
+
+    events = subscribed(/fail.*_claimed\.solid_queue/) do
+      SolidQueue::Process.prune
+    end
+
+    assert_equal 1, events.count
+    assert_event events.first, "fail_many_claimed", process_ids: [ process.id ], job_ids: jobs.map(&:id), size: 3
+  end
+
   test "errors when deregistering processes are included in deregister_process events" do
     previous_thread_report_on_exception, Thread.report_on_exception = Thread.report_on_exception, false
     error = RuntimeError.new("everything is broken")