Class: Silas::DeadJobRescuerJob

Inherits:
ActiveJob::Base
  • Object
show all
Defined in:
app/jobs/silas/dead_job_rescuer_job.rb

Overview

Mandatory durability infrastructure (proven in the Phase 0 spike): Solid Queue FAILS a dead worker's claimed jobs (ProcessExitError on supervisor reap, ProcessPrunedError on heartbeat prune) and nothing retries them — retry_on can't see queue-level errors. This job, run as a recurring task, is the recovery path; recovery latency ≈ alive_threshold + its cadence.

It also sweeps expired approvals while it's here.

Constant Summary collapse

DEAD_PROCESS_ERRORS = 
%w[
  SolidQueue::Processes::ProcessExitError
  SolidQueue::Processes::ProcessPrunedError
].freeze

Instance Method Summary collapse

Instance Method Details

#perform ⇒ Object 



17
18
19
 
# File 'app/jobs/silas/dead_job_rescuer_job.rb', line 17

def perform
  Silas.instrument(:rescue) { |payload| sweep(payload) }
end

#sweep(payload) ⇒ Object 



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'app/jobs/silas/dead_job_rescuer_job.rb', line 21

def sweep(payload)
  ToolInvocation.expire_stale!
  payload[:rescued] = 0
  payload[:stranded] = 0
  return 0 unless defined?(SolidQueue)

  rescued = 0
  SolidQueue::FailedExecution.includes(:job).find_each do |failed|
    if DEAD_PROCESS_ERRORS.include?(failed.error&.dig("exception_class"))
      failed.retry
      rescued += 1
    elsif failed.job&.class_name == "Silas::AgentLoopJob"
      payload[:stranded] += 1 if fail_stranded_turn(failed)
    end
  end
  payload[:rescued] = rescued
  rescued
end