Skip to content

Commit 174cb07

Browse files
rosaclaude
andcommitted
Protect the shutdown reap from release failures too
Extract the rescued release into attempt_to_release_claimed_jobs_by and use it from reap_terminated_forks as well: a database that is down when the supervisor shuts down should not abort the reap loop or skip exit hooks, for the same reason it should not stop a replacement. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent 0526401 commit 174cb07

2 files changed

Lines changed: 34 additions & 11 deletions

File tree

‎lib/solid_queue/fork_supervisor.rb‎

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -57,8 +57,7 @@ def reap_terminated_forks
5757
terminated_fork.mark_as_reaped
5858

5959
if !status.exited? || status.exitstatus.to_i > 0
60-
error = Processes::ProcessExitError.new(status)
61-
release_claimed_jobs_by(terminated_fork, with_error: error)
60+
attempt_to_release_claimed_jobs_by(terminated_fork, status)
6261
end
6362
end
6463

@@ -74,21 +73,23 @@ def replace_fork(pid, status)
7473
terminated_fork.mark_as_reaped
7574
payload[:fork] = terminated_fork
7675

77-
begin
78-
release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status))
79-
rescue StandardError => error
80-
# The database may be unreachable — likely the same reason the fork
81-
# terminated. Starting the replacement can't depend on it: the jobs
82-
# claimed by the terminated fork will be failed when its stale
83-
# registration is pruned once the database is back.
84-
handle_thread_error(error)
85-
end
76+
attempt_to_release_claimed_jobs_by(terminated_fork, status)
8677

8778
start_process(configured_processes.delete(pid))
8879
end
8980
end
9081
end
9182

83+
# The database may be unreachable — likely the same reason the fork
84+
# terminated. Neither starting a replacement nor shutting down can depend
85+
# on it: the jobs claimed by the terminated fork will be failed when its
86+
# stale registration is pruned once the database is back.
87+
def attempt_to_release_claimed_jobs_by(terminated_fork, status)
88+
release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status))
89+
rescue StandardError => error
90+
handle_thread_error(error)
91+
end
92+
9293
def all_processes_terminated?
9394
process_instances.empty?
9495
end

‎test/unit/fork_supervisor_test.rb‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,28 @@ def register
152152
end
153153
end
154154

155+
test "reap a terminated fork on shutdown even if releasing its claimed jobs fails" do
156+
configuration = SolidQueue::Configuration.new(workers: [ { queues: "*", processes: 1 } ], dispatchers: [], skip_recurring: true)
157+
supervisor = SolidQueue::ForkSupervisor.new(configuration)
158+
159+
terminated_fork = stub(kind: "Worker", name: "worker-42", hostname: "localhost", mark_as_reaped: true)
160+
supervisor.send(:process_instances)[42] = terminated_fork
161+
supervisor.send(:configured_processes)[42] = configuration.configured_processes.first
162+
163+
# The fork was killed and the database is unreachable when the supervisor
164+
# reaps it during shutdown
165+
status = stub(exited?: false, exitstatus: nil, pid: 42, signaled?: true, stopsig: nil, termsig: 9)
166+
::Process.stubs(:waitpid2).returns([ 42, status ]).then.returns(nil)
167+
supervisor.expects(:release_claimed_jobs_by).raises(ActiveRecord::ConnectionNotEstablished.new("connection is closed"))
168+
169+
assert_nothing_raised do
170+
supervisor.send(:reap_terminated_forks)
171+
end
172+
173+
assert_empty supervisor.send(:process_instances)
174+
assert_empty supervisor.send(:configured_processes)
175+
end
176+
155177
test "fail orphaned executions" do
156178
3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) }
157179
process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123")

0 commit comments

Comments
 (0)