Class: Plutonium::Interaction::Async::Executor

Inherits:
Object
  • Object
show all
Defined in:
lib/plutonium/interaction/async/executor.rb

Overview

Performs a run: resolves its targets, calls the subclass's work, keeps progress current, and applies the declared failure policy.

Separate from Run so the record stays a record, and separate from Job so the whole thing can be driven synchronously in a test or a console.

Authorization is re-derived here, per target, at the last moment

Context#targets answers "may the initiator act on these?" once, up front. That answer makes a good operator report but it is only true as of the moment it was computed, and a bulk run over thousands of records acts long afterwards. So the answer is re-derived immediately before each perform_on, because BOTH of its inputs go stale:

  • the SUBJECTS (initiator, tenant) are cached on the context. They are re-read on a clock — see SUBJECT_REFRESH_INTERVAL.
  • the RECORD is a snapshot from the resolution query, and predicates read record state (Blogging::PostPolicy#archive? is literally record.published?). Each target is re-read through the policy scope right before its check, which also catches a record that left the tenant mid-run.

A target that fails the re-check is RECORDED as a failure, never silently skipped: "you may no longer act on 3 of these" is exactly what tells an operator the run under-applied.

Resuming after an interruption

#call only ever STARTS from "pending" (see #claim!) — a run already "running" is left alone, since two concurrent executors on the same row would race. A run interrupted mid-batch (crash, dropped job) can still be resumed safely: reset it to "pending" (see Async::ReapJob) and re-enqueue. Context#targets resolves only Run#unhandled_target_ids, so a target already dispositioned before the interruption is not redone.

Defined Under Namespace

Classes: BatchAbortedError, TargetRefusedError

Constant Summary collapse

SUBJECT_REFRESH_INTERVAL =

How long a resolved (initiator, tenant) pair is trusted before Context#refresh_subjects! re-reads it.

Wall-clock rather than per-record because that is the shape of the risk being managed: revocation urgency is measured in seconds, not in records. Refreshing per record costs two queries per target — 20,000 extra queries on a 10,000-target run — to close a window this closes for a handful.

5.seconds

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(run) ⇒ Executor

Returns a new instance of Executor.



75
76
77
 
# File 'lib/plutonium/interaction/async/executor.rb', line 75

def initialize(run)
  @run = run
end

Instance Attribute Details

#run ⇒ Object (readonly)

Returns the value of attribute run.



73
74
75
 
# File 'lib/plutonium/interaction/async/executor.rb', line 73

def run
  @run
end

Instance Method Details

#call ⇒ void

This method returns an undefined value.



80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/plutonium/interaction/async/executor.rb', line 80

def call
  return unless claim!

  @context = build_context

  run.targeted? ? perform_targets : perform_opaque
rescue StandardError, NotImplementedError => e
  # Another executor owns this run now: ReapJob judged it stalled, reset
  # it to pending, and a second job claimed it — bumping lock_version out
  # from under us (see Async::ReapJob, #claim! and #superseded?).
  #
  # Returning without touching the row is the whole point. Every write
  # this executor still holds is stale by definition, so recording a
  # failure here would either raise again or overwrite the live
  # executor's progress with our older copy. The run is not failed; it is
  # simply no longer ours.
  #
  # Checked ahead of the failure path rather than in a rescue clause of
  # its own, because the two are told apart by the errored RECORD, not by
  # the exception class — see #superseded?.
  if superseded?(e)
    Rails.logger.warn {
      "plutonium: interaction run #{run.id} was reclaimed by another executor; abandoning this pass"
    }
    return
  end

  # NotImplementedError is NOT a StandardError, and two things here raise
  # it: a run subclass that implements no work at all, and a policy
  # predicate that has been renamed since enqueue (Policy#send_with_report).
  # Both must land in the run's log rather than escaping.
  #
  # Swallowed rather than re-raised because ActiveJob would retry, and a
  # retry re-applies every target the run already committed. The row is
  # the report, and it now reads failed, with the reason.
  Rails.logger.warn { "plutonium: interaction run #{run.id} (#{run.class}) failed: #{e.message}" }
  record_failure(e)
end