Class: GenevaDrive::Executor Private

Inherits:
Object
  • Object
show all
Defined in:
lib/geneva_drive/executor.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Executes a single step within a workflow context. Handles flow control signals, exception handling, and state transitions.

The Executor owns the step execution and workflow during execution, using pessimistic locking to ensure atomicity of state transitions.

Execution phases:

  1. Acquire locks, validate states, transition to executing/performing
  2. Release locks, execute user code (step block)
  3. Acquire locks, handle flow control result, transition to final states

Defined Under Namespace

Classes: Interruption

Constant Summary collapse

STEP_TRANSITIONS =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Valid state transitions for step executions

{
  "scheduled" => %w[scheduled waiting in_progress canceled skipped failed completed],
  "waiting" => %w[scheduled canceled skipped],
  "in_progress" => %w[in_progress completed failed canceled skipped]
}.freeze
WORKFLOW_TRANSITIONS =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Valid state transitions for workflows

{
  "ready" => %w[ready performing canceled paused finished],
  "performing" => %w[ready performing canceled paused finished]
}.freeze

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.execute!(step_execution, logger: nil, interruptible: true, max_iterations: nil) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Executes a step execution with full flow control and exception handling. Handles both regular steps and resumable (cursor-iterating) steps.

Examples:

Execute with a pre-tagged logger from a background job

logger = Rails.logger.tagged("job_id=#{job_id}")
GenevaDrive::Executor.execute!(step_execution, logger: logger)

Parameters:

  • step_execution (GenevaDrive::StepExecution) —

    the step to execute

  • logger (Logger, nil) (defaults to: nil) —

    optional base logger to inject into the workflow. If provided, this logger will be used as the base for all workflow logging, with workflow and step-specific tags added on top. This allows callers (background jobs, controllers, etc.) to pass in a logger that already has appropriate context tags (e.g., job_id, request_id).

  • interruptible (Boolean) (defaults to: true) —

    whether resumable steps respect interruption conditions (max_iterations, max_runtime, shutdown, external pause/cancel). Test helpers pass false to run a resumable step to completion in one call.

  • max_iterations (Integer, nil) (defaults to: nil) —

    overrides the step definition's max_iterations for this execution (used by the run_iterations test helper)



54
55
56
 
# File 'lib/geneva_drive/executor.rb', line 54

def self.execute!(step_execution, logger: nil, interruptible: true, max_iterations: nil)
  new.call(step_execution, logger: logger, interruptible: interruptible, max_iterations: max_iterations)
end

Instance Method Details

#call(step_execution, logger: nil, interruptible: true, max_iterations: nil) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Performs the step execution.

Parameters:

  • step_execution (GenevaDrive::StepExecution) —

    the step to execute

  • logger (Logger, nil) (defaults to: nil) —

    optional base logger to inject into the workflow

  • interruptible (Boolean) (defaults to: true) —

    whether resumable steps respect interruption conditions

  • max_iterations (Integer, nil) (defaults to: nil) —

    per-execution override of the step's max_iterations



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/geneva_drive/executor.rb', line 65

def call(step_execution, logger: nil, interruptible: true, max_iterations: nil)
  @step_execution = step_execution
  @workflow = step_execution.workflow
  @interruptible = interruptible
  @max_iterations_override = max_iterations

  # Build the full logger chain (base -> workflow -> step tags) and inject
  # it so step code calling `logger` gets the fully-tagged step execution
  # logger. Falls back to Rails.logger if no logger is provided.
  step_logger = build_step_logger(logger || Rails.logger, step_execution)
  Rails.error.set_context(geneva_drive: error_context)

  @workflow.with_logger(step_logger) do
    step_execution.with_logger(step_logger) do
      execute_with_logger(step_execution)
    end
  end
end

#should_interrupt?(iterations: nil) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Checks if a resumable step should be interrupted at a checkpoint. Called by IterableStep during checkpoint!.

Parameters:

  • iterations (Integer, nil) (defaults to: nil) —

    iterations completed so far in this execution

Returns:

  • (Boolean) 


89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/geneva_drive/executor.rb', line 89

def should_interrupt?(iterations: nil)
  # If interruption is disabled (e.g., in tests), never interrupt
  return false unless @interruptible

  max_iterations = @max_iterations_override || (@step_definition.respond_to?(:max_iterations) ? @step_definition.max_iterations : nil)
  return true if max_iterations && iterations && iterations >= max_iterations
  return true if max_runtime_exceeded?
  return true if job_should_exit?
  return true if workflow_interrupted?
  false
end