diff --git a/lib/solid_queue/fork_supervisor.rb b/lib/solid_queue/fork_supervisor.rb index 63595b00..00757afe 100644 --- a/lib/solid_queue/fork_supervisor.rb +++ b/lib/solid_queue/fork_supervisor.rb @@ -57,8 +57,7 @@ def reap_terminated_forks terminated_fork.mark_as_reaped if !status.exited? || status.exitstatus.to_i > 0 - error = Processes::ProcessExitError.new(status) - release_claimed_jobs_by(terminated_fork, with_error: error) + attempt_to_release_claimed_jobs_by(terminated_fork, status) end end @@ -73,14 +72,24 @@ def replace_fork(pid, status) if terminated_fork = process_instances.delete(pid) terminated_fork.mark_as_reaped payload[:fork] = terminated_fork - error = Processes::ProcessExitError.new(status) - release_claimed_jobs_by(terminated_fork, with_error: error) + + attempt_to_release_claimed_jobs_by(terminated_fork, 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/test/unit/fork_supervisor_test.rb b/test/unit/fork_supervisor_test.rb index 519e4f88..0d809227 100644 --- a/test/unit/fork_supervisor_test.rb +++ b/test/unit/fork_supervisor_test.rb @@ -129,6 +129,51 @@ def register terminate_process(pid) end + test "replace a terminated fork even if releasing its claimed jobs fails" do + configuration = SolidQueue::Configuration.new(workers: [ { queues: "*", processes: 1 } ], dispatchers: [], skip_recurring: true) + supervisor = SolidQueue::ForkSupervisor.new(configuration) + + configured_process = configuration.configured_processes.first + terminated_fork = stub(kind: "Worker", name: "worker-42", hostname: "localhost", mark_as_reaped: true) + + supervisor.send(:process_instances)[42] = terminated_fork + supervisor.send(:configured_processes)[42] = configured_process + + # The database is unreachable when the supervisor tries to fail the + # terminated fork's claimed jobs, like right after a database restart + # that also crashed the fork + supervisor.expects(:release_claimed_jobs_by).raises(ActiveRecord::ConnectionNotEstablished.new("connection is closed")) + supervisor.expects(:start_process).with(configured_process) + + status = stub(exited?: true, exitstatus: 1, pid: 42, signaled?: false, stopsig: nil, termsig: nil) + + assert_nothing_raised do + supervisor.send(:replace_fork, 42, status) + end + end + + test "reap a terminated fork on shutdown even if releasing its claimed jobs fails" do + configuration = SolidQueue::Configuration.new(workers: [ { queues: "*", processes: 1 } ], dispatchers: [], skip_recurring: true) + supervisor = SolidQueue::ForkSupervisor.new(configuration) + + terminated_fork = stub(kind: "Worker", name: "worker-42", hostname: "localhost", mark_as_reaped: true) + supervisor.send(:process_instances)[42] = terminated_fork + supervisor.send(:configured_processes)[42] = configuration.configured_processes.first + + # The fork was killed and the database is unreachable when the supervisor + # reaps it during shutdown + status = stub(exited?: false, exitstatus: nil, pid: 42, signaled?: true, stopsig: nil, termsig: 9) + ::Process.stubs(:waitpid2).returns([ 42, status ]).then.returns(nil) + supervisor.expects(:release_claimed_jobs_by).raises(ActiveRecord::ConnectionNotEstablished.new("connection is closed")) + + assert_nothing_raised do + supervisor.send(:reap_terminated_forks) + end + + assert_empty supervisor.send(:process_instances) + assert_empty supervisor.send(:configured_processes) + end + test "fail orphaned executions" do 3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) } process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123")