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