From 0f52e426ad5708e603959d99bd3fb4639e6d803e Mon Sep 17 00:00:00 2001 From: wintan1418 Date: Tue, 18 Aug 2026 20:03:24 +0100 Subject: [PATCH] Retry finalizing claimed executions on transient errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a job finishes, the worker deletes its claimed execution and marks the job finished (or failed) in a small transaction. If that transaction raises — say a transient DB connection drop — the error is swallowed by the pool's thread error handler and the claimed execution stays claimed forever: every recovery mechanism (releasing on deregistration, failing orphaned or pruned claims) assumes the claiming process is gone, but this worker is alive and well, so nothing ever picks the execution up again, even though the job already ran to completion. Finalization is idempotent — it locks the claimed execution row and no-ops if it's already gone — so it's safe to retry. Wait out the hiccup with a few increasingly spaced attempts before giving up and letting the error propagate as before. Each retry emits a retry_finalization.solid_queue event, logged at warn level, so operators can see the instability. Fixes #748 --- app/models/solid_queue/claimed_execution.rb | 29 ++++++++- lib/solid_queue/log_subscriber.rb | 7 +++ .../solid_queue/claimed_execution_test.rb | 63 +++++++++++++++++++ test/unit/log_subscriber_test.rb | 7 +++ 4 files changed, 104 insertions(+), 2 deletions(-) diff --git a/app/models/solid_queue/claimed_execution.rb b/app/models/solid_queue/claimed_execution.rb index 9a606e0b..24166b46 100644 --- a/app/models/solid_queue/claimed_execution.rb +++ b/app/models/solid_queue/claimed_execution.rb @@ -5,6 +5,8 @@ class SolidQueue::ClaimedExecution < SolidQueue::Execution scope :orphaned, -> { where.missing(:process) } + FINALIZATION_RETRY_DELAYS = [ 0.5.seconds, 1.second, 2.seconds, 4.seconds ] + class Result < Struct.new(:success, :error) def success? success @@ -66,9 +68,9 @@ def perform result = execute if result.success? - finished + retrying_finalization { finished } else - failed_with(result.error) + retrying_finalization { failed_with(result.error) } raise result.error end end @@ -102,6 +104,29 @@ def finished finalize { job.finished! } end + # The job has already run by the time we record its outcome. If recording it + # fails (e.g. a transient DB error), the execution would stay claimed forever: + # claimed executions are only recovered when their process is gone, and this + # one's worker is still alive. Finalization is idempotent, so wait out the + # hiccup and try again before giving up. + def retrying_finalization(&block) + attempts = 0 + + begin + block.call + rescue StandardError => error + if delay = FINALIZATION_RETRY_DELAYS[attempts] + attempts += 1 + SolidQueue.instrument(:retry_finalization, job_id: job_id, process_id: process_id, attempt: attempts, error: error) + + sleep(delay) + retry + else + raise + end + end + end + def finalize finalized = unless_already_finalized do yield diff --git a/lib/solid_queue/log_subscriber.rb b/lib/solid_queue/log_subscriber.rb index 4bb52baf..89527ef2 100644 --- a/lib/solid_queue/log_subscriber.rb +++ b/lib/solid_queue/log_subscriber.rb @@ -26,6 +26,13 @@ def release_claimed(event) info formatted_event(event, action: "Release claimed job", **event.payload.slice(:job_id, :process_id)) end + def retry_finalization(event) + attributes = event.payload.slice(:job_id, :process_id, :attempt) + attributes[:error] = formatted_error(event.payload[:error]) if event.payload[:error] + + warn formatted_event(event, action: "Retry finalizing claimed job", **attributes) + end + def retry_all(event) debug formatted_event(event, action: "Retry failed jobs", **event.payload.slice(:jobs_size, :size)) end diff --git a/test/models/solid_queue/claimed_execution_test.rb b/test/models/solid_queue/claimed_execution_test.rb index 72a988f6..ed1aa290 100644 --- a/test/models/solid_queue/claimed_execution_test.rb +++ b/test/models/solid_queue/claimed_execution_test.rb @@ -59,6 +59,69 @@ class SolidQueue::ClaimedExecutionTest < ActiveSupport::TestCase assert_equal @process, claimed_execution.process end + test "finishing a job is retried when recording the outcome fails transiently" do + claimed_execution = prepare_and_claim_job AddToBufferJob.perform_later(42) + claimed_execution.stubs(:sleep) + job = claimed_execution.job + + attempts = 0 + job.define_singleton_method(:finished!) do + attempts += 1 + raise ActiveRecord::ConnectionNotEstablished if attempts == 1 + super() + end + + assert_difference -> { SolidQueue::ClaimedExecution.count }, -1 do + claimed_execution.perform + end + + assert_equal 2, attempts + assert job.reload.finished? + end + + test "failing a job is retried when recording the outcome fails transiently" do + claimed_execution = prepare_and_claim_job RaisingJob.perform_later(RuntimeError, "A") + claimed_execution.stubs(:sleep) + job = claimed_execution.job + + attempts = 0 + job.define_singleton_method(:failed_with) do |error| + attempts += 1 + raise ActiveRecord::ConnectionNotEstablished if attempts == 1 + super(error) + end + + assert_difference -> { SolidQueue::ClaimedExecution.count } => -1, -> { SolidQueue::FailedExecution.count } => 1 do + assert_raises RuntimeError do + claimed_execution.perform + end + end + + assert_equal 2, attempts + assert job.reload.failed? + end + + test "finalization failures are raised once retries are exhausted" do + claimed_execution = prepare_and_claim_job AddToBufferJob.perform_later(42) + claimed_execution.stubs(:sleep) + job = claimed_execution.job + + attempts = 0 + job.define_singleton_method(:finished!) do + attempts += 1 + raise ActiveRecord::ConnectionNotEstablished + end + + assert_no_difference -> { SolidQueue::ClaimedExecution.count } do + assert_raises ActiveRecord::ConnectionNotEstablished do + claimed_execution.perform + end + end + + assert_equal SolidQueue::ClaimedExecution::FINALIZATION_RETRY_DELAYS.size + 1, attempts + assert_not job.reload.finished? + end + test "job failures are reported via Rails error subscriber" do subscriber = ErrorBuffer.new diff --git a/test/unit/log_subscriber_test.rb b/test/unit/log_subscriber_test.rb index 9e239923..02a3b7d6 100644 --- a/test/unit/log_subscriber_test.rb +++ b/test/unit/log_subscriber_test.rb @@ -59,6 +59,13 @@ def set_logger(logger) assert_match_logged :warn, "Terminate Worker that failed to boot in time", "pid: 42, hostname: \"#{worker.hostname}\", name: \"#{worker.name}\"" end + test "retry finalizing claimed job" do + attach_log_subscriber + instrument "retry_finalization.solid_queue", job_id: 42, process_id: 43, attempt: 1, error: ActiveRecord::ConnectionNotEstablished.new("connection lost") + + assert_match_logged :warn, "Retry finalizing claimed job", "job_id: 42, process_id: 43, attempt: 1, error: \"ActiveRecord::ConnectionNotEstablished connection lost\"" + end + test "deregister process" do process = SolidQueue::Process.register(kind: "Worker", pid: 42, hostname: "localhost", name: "worker-123") last_heartbeat_at = process.last_heartbeat_at.iso8601