diff --git a/lib/solid_queue/async_supervisor.rb b/lib/solid_queue/async_supervisor.rb index 4a7f4aeec..f6ab38342 100644 --- a/lib/solid_queue/async_supervisor.rb +++ b/lib/solid_queue/async_supervisor.rb @@ -27,8 +27,7 @@ def replace_thread(thread_id) if (instance = process_instances.delete(thread_id)) payload[:thread] = instance - error = Processes::ThreadTerminatedError.new(instance.name) - release_claimed_jobs_by(instance, with_error: error) + attempt_to_release_claimed_jobs_by(instance, with_error: Processes::ThreadTerminatedError.new(instance.name)) start_process(configured_processes.delete(thread_id)) end diff --git a/lib/solid_queue/fork_supervisor.rb b/lib/solid_queue/fork_supervisor.rb index 00757afe2..0acc36e7d 100644 --- a/lib/solid_queue/fork_supervisor.rb +++ b/lib/solid_queue/fork_supervisor.rb @@ -57,7 +57,7 @@ def reap_terminated_forks terminated_fork.mark_as_reaped if !status.exited? || status.exitstatus.to_i > 0 - attempt_to_release_claimed_jobs_by(terminated_fork, status) + attempt_to_release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status)) end end @@ -73,23 +73,13 @@ def replace_fork(pid, status) terminated_fork.mark_as_reaped payload[:fork] = terminated_fork - attempt_to_release_claimed_jobs_by(terminated_fork, status) + attempt_to_release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status)) start_process(configured_processes.delete(pid)) end end end - # The database may be unreachable — likely the same reason the fork - # terminated. Neither starting a replacement nor shutting down can depend - # on it: the jobs claimed by the terminated fork will be failed when its - # stale registration is pruned once the database is back. - def attempt_to_release_claimed_jobs_by(terminated_fork, status) - release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status)) - rescue StandardError => error - handle_thread_error(error) - end - def all_processes_terminated? process_instances.empty? end diff --git a/lib/solid_queue/supervisor/maintenance.rb b/lib/solid_queue/supervisor/maintenance.rb index d92569d58..23e321f47 100644 --- a/lib/solid_queue/supervisor/maintenance.rb +++ b/lib/solid_queue/supervisor/maintenance.rb @@ -43,5 +43,15 @@ def release_claimed_jobs_by(terminated_process, with_error:) end end end + + # The database may be unreachable — possibly the same reason the + # supervised process died. Neither starting a replacement nor shutting + # down can depend on it: the jobs claimed by the dead process will be + # failed when its stale registration is pruned once the database is back. + def attempt_to_release_claimed_jobs_by(terminated_process, with_error:) + release_claimed_jobs_by(terminated_process, with_error: with_error) + rescue StandardError => error + handle_thread_error(error) + end end end diff --git a/test/unit/async_supervisor_test.rb b/test/unit/async_supervisor_test.rb index 0ca6a3c2f..f2a70f0ff 100644 --- a/test/unit/async_supervisor_test.rb +++ b/test/unit/async_supervisor_test.rb @@ -112,6 +112,26 @@ class AsyncSupervisorTest < ActiveSupport::TestCase assert_no_match /the database connection pool is/, log.string end + test "replace a terminated thread even if releasing its claimed jobs fails" do + configuration = SolidQueue::Configuration.new(workers: [ { queues: "*", processes: 1 } ], dispatchers: [], skip_recurring: true) + supervisor = SolidQueue::AsyncSupervisor.new(configuration) + + configured_process = configuration.configured_processes.first + terminated_thread = stub(kind: "Worker", name: "worker-42", hostname: "localhost", alive?: false) + + supervisor.send(:process_instances)[42] = terminated_thread + supervisor.send(:configured_processes)[42] = configured_process + + # The database is unreachable when the supervisor tries to fail the + # terminated thread's claimed jobs, like right after losing its connection + supervisor.expects(:release_claimed_jobs_by).raises(ActiveRecord::ConnectionNotEstablished.new("connection is closed")) + supervisor.expects(:start_process).with(configured_process) + + assert_nothing_raised do + supervisor.send(:check_and_replace_terminated_processes) + end + end + private def run_supervisor_as_thread(**options) SolidQueue::Supervisor.start(mode: :async, standalone: false, **options.with_defaults(skip_recurring: true))