Class: Plutonium::Interaction::Async::Job

Inherits:
ActiveJob::Base
  • Object
show all
Defined in:
lib/plutonium/interaction/async/job.rb

Overview

The ActiveJob entry point for a run.

The only thing that crosses the process boundary is the run's id. Every scrap of context — who started it, which tenant, which targets, which policy — is re-read from the row by Context, so nothing is inherited from the dispatching request. That is what makes a run safe to perform minutes or hours later, on another machine.

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.concurrency_duration ⇒ Object

Read at dispatch time rather than passed to limits_concurrency here, for the same reason queue_as takes a block — this class is autoloaded, and a host configures stall_after in an initializer.

stall_after is the right duration because it is the same question: "how long may a run be silent before we assume its worker is dead?" The semaphore then expires no later than the point the reaper would resume the run anyway. Solid Queue's 3-minute default would expire mid-batch on any run big enough to be worth dispatching, handing the exclusivity away while the work is still going.



56
 
# File 'lib/plutonium/interaction/async/job.rb', line 56

def self.concurrency_duration = Plutonium.configuration.async_interactions.stall_after

Instance Method Details

#perform(run_id) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/plutonium/interaction/async/job.rb', line 59

def perform(run_id)
  run = Plutonium::Interaction::Async::Run.find_by(id: run_id)

  # A run deleted between enqueue and perform is not an error — there is
  # simply nothing to do, and raising would only retry until the queue
  # gives up.
  return if run.nil?

  # Idempotence, and the guard against a retry re-applying committed
  # work: a settled run has already reported its outcome, and performing
  # it again would act on its targets a second time.
  unless run.in_progress?
    Rails.logger.warn { "plutonium: interaction run #{run.id} is #{run.state}; skipping" }
    return
  end

  Executor.new(run).call
end