Retry finalizing claimed executions on transient errors

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

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

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
 

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