Class: Plutonium::Interaction::Async::Run

Inherits:
ActiveRecord::Base
  • Object
show all
Includes:
Resource::Record
Defined in:
lib/plutonium/interaction/async/run.rb

Overview

A persisted interaction run.

STI base: authors subclass to define execution and failure policy, and the subclass name lands in type. Options are JSON, so an async action costs no migration.

This class deliberately holds NO execution logic — that lives in the runs executor. It is the record; the executor is the behaviour. Keeping them apart is what lets the job rebuild an authorization context around the work without the model knowing anything about policies.

Constant Summary collapse

STATES = 
%w[pending running completed failed].freeze
IN_PROGRESS_STATES = 
%w[pending running].freeze
FAILURE_POLICIES = 
%i[halt continue transactional].freeze
MODEL_NAME =

Routes, paths and helpers are all derived from model_name, and the default would spell the gem's own namespace out in every URL (/admin/plutonium/interaction/runs) and every helper (plutonium_async_run_path). Pinned to the BASE class rather than self so every STI subclass routes to the one registered resource: resource_url_for(a TestPostRun) has to find the Run's route config.

"Run" rather than the shorter "Run" on purpose: a bare "Run" would squat the most generic name available and collide silently with a host app's own Run model (CI runs, ML training runs, delivery runs) on every axis that matters -- route path, controller name, param_key and i18n key are all identical, and nothing in Resource::Register detects it.

ActiveModel::Name.new(self, nil, "AsyncRun")

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Models::HasCents

#has_cents_unit_for

Class Method Details

.model_name ⇒ Object 



44
 
# File 'lib/plutonium/interaction/async/run.rb', line 44

def self.model_name = MODEL_NAME

.on_failure(policy) ⇒ void

This method returns an undefined value.

Declares how the executor reacts to a target failure. Rejects an unknown policy at class-definition time rather than letting a typo surface as surprise behaviour deep inside a background job.

Parameters:



101
102
103
104
105
106
107
108
 
# File 'lib/plutonium/interaction/async/run.rb', line 101

def self.on_failure(policy)
  unless FAILURE_POLICIES.include?(policy)
    raise ArgumentError,
      "unknown failure policy: #{policy.inspect} (expected one of #{FAILURE_POLICIES.map(&:inspect).join(", ")})"
  end

  self.failure_policy = policy
end

Instance Method Details

#attachment(key) ⇒ Plutonium::Attachments::Resolved?

An attachment attribute, revived from the token dispatch staged.

A file cannot ride the options column, so dispatch uploads it to the backend's cache and stores the token (see Dispatchable#stage_dispatch_attachments). options["file"] is therefore that token — a String — and this is what turns it back into something with a filename and bytes.

async do
def perform
  CSV.foreach(attachment(:import_file).url) { |row| ... }
end
end

Deliberately not folded into options. The token is what is actually stored, and reviving it reaches storage — which the progress page reads options without wanting to do.

Parameters:

  • key (Symbol, String) —

    the attribute name

Returns:



279
 
# File 'lib/plutonium/interaction/async/run.rb', line 279

def attachment(key) = attachments(key).first

#attachments(key) ⇒ Array<Plutonium::Attachments::Resolved>

Every attachment staged under key, for a multiple-file attribute.

Parameters:

  • key (Symbol, String)

Returns:



285
 
# File 'lib/plutonium/interaction/async/run.rb', line 285

def attachments(key) = Plutonium::Attachments.resolve(options[key.to_s])

#error_count ⇒ Integer

Returns recorded target failures, run-level entries included.

Returns:

  • (Integer) —

    recorded target failures, run-level entries included



137
 
# File 'lib/plutonium/interaction/async/run.rb', line 137

def error_count = errors_log.size

#fail!(message = nil) ⇒ Object 



114
115
116
117
 
# File 'lib/plutonium/interaction/async/run.rb', line 114

def fail!(message = nil)
  record_target_failure!(id: nil, message: message) if message
  update!(state: "failed", finished_at: Time.current, last_activity_at: Time.current)
end

#finish! ⇒ Object 



112
 
# File 'lib/plutonium/interaction/async/run.rb', line 112

def finish! = update!(state: "completed", finished_at: Time.current, last_activity_at: Time.current)

#heartbeat! ⇒ Time

Says "still working" — for work the executor cannot see inside.

Async::ReapJob treats a run silent past stall_after as dead and resumes it. Every write the executor makes refreshes that clock, so a targeted run whose targets are quick heartbeats once per target for free. Two shapes of work get nothing:

  • OPAQUE work. Between the claim and #finish! the executor writes nothing at all, because there is nothing to count. An opaque perform that outlives stall_after is reaped mid-flight, and — having no handled_target_ids to resume from — is re-run FROM SCRATCH.
  • A single perform_on that outlives stall_after on its own.

So a run that does either must say so itself. This is deliberately not automatic: a background thread would have to guess how often to write, and would keep reporting a wedged worker as healthy. Only the work knows it is making progress.

def perform
invoices.each_slice(500) do |slice|
  reissue(slice)
  heartbeat!
end
end

It also ANSWERS: the write is conditional on this instance still holding the row's lock_version, so a worker superseded while inside a long perform learns at its next beat rather than at the end — see Async::Executor#superseded?, which turns that into "no longer mine" instead of a failure. That is the only place a long opaque run can find out at all.

The beat must NOT bump lock_version — that is the executor's own record it would be invalidating, making its next save! raise against a row nobody else touched. Which rules out more than it looks like:

  • touch and update! both bump it (Locking::Optimistic hooks _touch_row as well as _update_record).
  • so does update_all GIVEN A HASH. Rails silently adds the increment when the model locks and the hash does not mention the column — see ActiveRecord::Relation#update_all. Only the string/array form escapes it, going through sanitize_sql_for_assignment untouched, which is why this and Async::Executor#claim! both spell their SQL out.

The in-memory attribute is left alone for a related reason: nothing reads it, and assigning it would put a change on the record that the caller never asked to save.

Call it on SELF, which is what +perform+/+perform_on+ already hand the author. The conditional reads this instance's lock_version, and the executor's own writes keep that current; a separately loaded copy (Run.find(id)) goes stale at the first advance! and would raise here having been superseded by nobody.

Under the :transactional policy the beat is inside the batch's transaction like everything else, so it is invisible to the reaper until the batch commits. An all-or-nothing batch longer than stall_after is reapable no matter what this does.

Returns:

  • (Time) —

    the recorded activity time

Raises:

  • (ActiveRecord::StaleObjectError) —

    if another executor owns the run



348
349
350
351
352
353
354
355
 
# File 'lib/plutonium/interaction/async/run.rb', line 348

def heartbeat!
  now = Time.current
  written = self.class.where(id: id, lock_version: lock_version)
    .update_all(["last_activity_at = ?", now])
  raise ActiveRecord::StaleObjectError.new(self, "heartbeat") if written.zero?

  now
end

#in_progress? ⇒ Boolean

Returns:

  • (Boolean) 


119
 
# File 'lib/plutonium/interaction/async/run.rb', line 119

def in_progress? = IN_PROGRESS_STATES.include?(state)

#options ⇒ Hash

The interaction's validated inputs, with the types dispatch put in.

The column is JSON, so Dispatchable#dispatch_options writes it through ActiveJob::Arguments; this is the other half. Without it a Date comes back a String and a BigDecimal comes back a String, which is the kind of bug that survives every test written against a run whose options happen to be strings anyway.

Falls back to the raw hash for a row written before this existed, and for one an operator edited by hand — neither carries the envelope keys deserialize expects, and refusing to read them would take out the progress page as well as the work.

Returns:

  • (Hash) 


249
250
251
252
253
254
255
256
 
# File 'lib/plutonium/interaction/async/run.rb', line 249

def options
  raw = super
  return raw unless raw.is_a?(Hash)

  ::ActiveJob::Arguments.deserialize([raw]).first
rescue ::ActiveJob::DeserializationError
  raw
end

#outcome ⇒ String

What actually happened, as opposed to how the executor exited.

A :continue run that could not apply some of its targets ends as completed (see Async::Executor#perform_targets) — the author declared partial application acceptable and the loop ran to the end. But "completed" on its own is what a clean success looks like, so rendering it alone would make a run that under-applied indistinguishable from one that did everything asked of it. Every reader — badge, table column, progress panel — goes through this instead of through state.

Returns:

  • (String) 


132
133
134
 
# File 'lib/plutonium/interaction/async/run.rb', line 132

def outcome
  (state == "completed" && errors_log.any?) ? "completed_with_errors" : state
end

#perform ⇒ Object

Opaque (untargeted) work. Subclasses override this, or perform_on for per-target work.

Defined here so a subclass that implements NEITHER is diagnosed by name rather than surfacing as a NoMethodError from inside the executor's loop, where nothing in the message says which class is at fault.

Raises:

  • (NotImplementedError) 


374
375
376
377
378
 
# File 'lib/plutonium/interaction/async/run.rb', line 374

def perform
  raise NotImplementedError,
    "#{self.class} implements neither #perform nor #perform_on: define " \
    "#perform_on(record) for work over targets, or #perform for opaque work"
end

#progress_fraction ⇒ Float?

nil means INDETERMINATE, not zero: opaque work has no denominator, and the progress UI renders a spinner rather than a 0% bar.

Returns:

  • (Float, nil) 


179
180
181
182
183
 
# File 'lib/plutonium/interaction/async/run.rb', line 179

def progress_fraction
  return nil if progress_total.nil? || progress_total.zero?

  progress_done.to_f / progress_total
end

#record_target_failure!(id:, message:) ⇒ Object

Appends a failure to errors_log and persists immediately, preserving prior entries. It writes rather than staging because the information it carries — which targets the run could not act on — must survive an early return, a rescue that never reaches #fail!, or a caller that simply forgets to save. A silently-dropped entry is indistinguishable from a clean run, which is the one thing an operator must never be told.

A nil id is the RUN-LEVEL sentinel: the failure is the whole run's (see #fail!), not one target's — it advances neither progress_done nor handled_target_ids. Readers grouping the log by target must treat nil as its own bucket rather than a target id.



196
197
198
 
# File 'lib/plutonium/interaction/async/run.rb', line 196

def record_target_failure!(id:, message:)
  record_target_failures!([{id: id, message: message}])
end

#record_target_failures!(entries) ⇒ Object

Appends several failures in ONE write — folding in the progress bump and handled_target_ids for every entry that names a real target, so a crash between separate writes can't leave them out of sync with each other.

The singular form persists on every call, so a loop over M ids is M writes, each rewriting the whole errors_log JSON — O(M²) bytes. The executor resolves every unavailable target in one pass before it performs anything (see Async::Executor#record_unresolved), and that batch is what this exists for.

Parameters:

  • entries (Array<Hash>) —

    {id:, message:} pairs



212
213
214
215
216
217
218
219
220
221
222
223
 
# File 'lib/plutonium/interaction/async/run.rb', line 212

def record_target_failures!(entries)
  return if entries.empty?

  target_entries = entries.reject { |entry| entry[:id].nil? }
  appended = entries.map { |entry| {"target_id" => entry[:id], "message" => entry[:message]} }
  update!(
    errors_log: errors_log + appended,
    progress_done: progress_done + target_entries.size,
    handled_target_ids: handled_target_ids + target_entries.map { |entry| entry[:id].to_s },
    last_activity_at: Time.current
  )
end

#start! ⇒ Object 



110
 
# File 'lib/plutonium/interaction/async/run.rb', line 110

def start! = update!(state: "running", started_at: Time.current, last_activity_at: Time.current)

#target_label ⇒ String?

Human, I18n-aware name of the target resource class — "Post", not "Blogging::Post" — for display (see Async::RunDefinition). Falls back to the raw string for a target_type renamed/removed since this run was dispatched, rather than raising on an old row. nil for opaque (untargeted) work.

Returns:

  • (String, nil) 


146
147
148
149
150
151
152
 
# File 'lib/plutonium/interaction/async/run.rb', line 146

def target_label
  return nil if target_type.nil?

  target_type.constantize.model_name.human
rescue NameError
  target_type
end

#targeted? ⇒ Boolean

Which shape of work this is, decided by what the subclass implements rather than a mode flag — one less thing for an author to keep in sync.

Non-public methods count. private def perform_on(record) is a natural idiom for work only the executor is meant to invoke, and Ruby's public-only respond_to? default would read that as opaque work — routing it to #perform, which the base class raises NotImplementedError for. The author would see every dispatch fail on a run whose perform_on is right there. Async::Executor invokes both through send to match.

Returns:

  • (Boolean) 


366
 
# File 'lib/plutonium/interaction/async/run.rb', line 366

def targeted? = respond_to?(:perform_on, true)

#to_label ⇒ String

Runs have no name or title, so Labeling would fall back to "Async run #12" — true but silent about the only thing that distinguishes one row from the next.

Demodulized: a host's run class is as likely to be Billing::ReissueInvoicesRun as a top-level one, and the namespace adds nothing to a label that already sits under the run's own breadcrumb.

Except when demodulizing is what throws the name away. A run declared by Dispatchable#async is Blogging::ArchivePosts::Run, and every one of them would render "Run #12" — the failure mode this method exists to avoid, reintroduced for the shape most authors will write. For those, the enclosing segment IS the name, so it is folded back in.

Returns:

  • (String) 


169
170
171
172
173
 
# File 'lib/plutonium/interaction/async/run.rb', line 169

def to_label
  parts = self.class.name.split("::")
  name = (parts.last == "Run" && parts.size > 1) ? "#{parts[-2]}Run" : parts.last
  "#{name.titleize} ##{to_param}"
end

#unhandled_target_ids ⇒ Array

Target ids not yet dispositioned — what a resumed run still has to do. See #record_target_failures! and Async::Executor#advance!, which are what populate handled_target_ids.

Returns:

  • (Array) 


230
231
232
233
 
# File 'lib/plutonium/interaction/async/run.rb', line 230

def unhandled_target_ids
  handled = handled_target_ids.map(&:to_s).to_set
  target_ids.reject { |id| handled.include?(id.to_s) }
end