Class: GenevaDrive::StepExecution

Inherits:
ActiveRecord::Base
  • Object
show all
Includes:
MetadataAccessor
Defined in:
lib/geneva_drive/step_execution.rb

Overview

Represents a single step execution attempt within a workflow. Serves as both an audit record and an idempotency key.

Each step execution tracks:

  • The step being executed
  • The current state of execution
  • When the step was scheduled, started, and completed
  • Any errors that occurred
  • The outcome of the execution

Examples:

Querying step executions

workflow.step_executions.completed.each do |exec|
  puts "#{exec.step_name}: #{exec.outcome}"
end

Defined Under Namespace

Modules: MetadataAccessor

Constant Summary collapse

OUTCOMES =

Outcome values for audit purposes

%w[
  success
  continued
  reattempted
  skipped
  canceled
  failed
  recovered
  workflow_paused
].freeze

Class Method Summary collapse

Instance Method Summary collapse

Methods included from MetadataAccessor

#exception_info, #read_metadata, #reattempt_reason, #write_metadata

Class Method Details

.reset_resumable_columns_cache! ⇒ void

This method returns an undefined value.

Clears the cached detection result. Call this in tests or after running migrations in-process so the next access re-checks.



110
111
112
 
# File 'lib/geneva_drive/step_execution.rb', line 110

def reset_resumable_columns_cache!
  remove_instance_variable(:@_resumable_columns) if defined?(@_resumable_columns)
end

.reset_signal_columns_cache! ⇒ void

This method returns an undefined value.

Clears the cached signal column detection result. Call this in tests or after running migrations in-process so the next access re-checks.



137
138
139
 
# File 'lib/geneva_drive/step_execution.rb', line 137

def reset_signal_columns_cache!
  remove_instance_variable(:@_signal_columns) if defined?(@_signal_columns)
end

.resumable_columns? ⇒ Boolean

Lazily checks whether the resumable-step columns (cursor and continues_from_id) have been migrated. Never hits the database at class definition time — only on the first runtime call.

Deployments usually ship the gem update before running migrations, so all cursor/chaining behavior must degrade safely when the columns are absent: reads return nil, writes are no-ops, and executing an actual resumable_step raises a clear configuration error instead.

Returns:

  • (Boolean) 


96
97
98
99
100
101
102
103
104
 
# File 'lib/geneva_drive/step_execution.rb', line 96

def resumable_columns?
  if defined?(@_resumable_columns)
    return @_resumable_columns
  end

  @_resumable_columns = table_exists? &&
    column_names.include?("cursor") &&
    column_names.include?("continues_from_id")
end

.serialize_cursor(value) ⇒ Object?

Serializes a cursor value using ActiveJob serializers (handles Date, Time, and other types) and enforces GenevaDrive.max_cursor_size on the serialized JSON. The single serialization path for cursor writes.

Parameters:

  • value (Object, nil) —

    the cursor value

Returns:

  • (Object, nil) —

    the serialized cursor

Raises:



148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/geneva_drive/step_execution.rb', line 148

def serialize_cursor(value)
  return nil if value.nil?

  serialized = ActiveJob::Arguments.serialize([value]).first

  limit = GenevaDrive.max_cursor_size
  if limit
    bytesize = JSON.generate(serialized).bytesize
    if bytesize > limit
      raise GenevaDrive::CursorTooLargeError,
        "Serialized cursor is #{bytesize} bytes, exceeding GenevaDrive.max_cursor_size " \
        "(#{limit} bytes). The cursor is a position marker (an id, page number, or token), " \
        "not a place to store the data being processed. Set GenevaDrive.max_cursor_size to " \
        "nil to disable this check."
    end
  end

  serialized
end

.signal_columns? ⇒ Boolean

Lazily checks whether the signal columns (signal_id and waiting_since) have been migrated, mirroring resumable_columns?. Never hits the database at class definition time.

Everything unrelated to signals keeps working without them; a step declaring wait_for: fails loudly with a "run the install generator" message instead of silently running without its event.

Returns:

  • (Boolean) 


123
124
125
126
127
128
129
130
131
 
# File 'lib/geneva_drive/step_execution.rb', line 123

def signal_columns?
  if defined?(@_signal_columns)
    return @_signal_columns
  end

  @_signal_columns = table_exists? &&
    column_names.include?("signal_id") &&
    column_names.include?("waiting_since")
end

Instance Method Details

#cursor_value ⇒ Object?

Returns the deserialized cursor value for resumable steps. Uses ActiveJob serializers to handle Date, Time, and other types. Returns nil when the cursor column has not been migrated yet.

Returns:

  • (Object, nil) —

    the cursor value



250
251
252
253
254
 
# File 'lib/geneva_drive/step_execution.rb', line 250

def cursor_value
  return nil unless self.class.resumable_columns?
  return nil if cursor.blank?
  ActiveJob::Arguments.deserialize([cursor]).first
end

#cursor_value=(value) ⇒ void

This method returns an undefined value.

Sets the cursor value for resumable steps. Uses ActiveJob serializers to handle Date, Time, and other types. Silent no-op when the cursor column has not been migrated yet.

Parameters:

  • value (Object) —

    the cursor value to store

Raises:



263
264
265
266
 
# File 'lib/geneva_drive/step_execution.rb', line 263

def cursor_value=(value)
  return unless self.class.resumable_columns?
  self.cursor = self.class.serialize_cursor(value)
end

#execute!(interruptible: true, max_iterations: nil) ⇒ void

This method returns an undefined value.

Executes this step using the Executor.

Parameters:

  • 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



288
289
290
 
# File 'lib/geneva_drive/step_execution.rb', line 288

def execute!(interruptible: true, max_iterations: nil)
  GenevaDrive::Executor.execute!(self, interruptible: interruptible, max_iterations: max_iterations)
end

#logger ⇒ Logger

Same as ActiveRecord::Base#logger but supplemented with tags for step and workflow

Returns:

  • (Logger) 


295
296
297
298
299
300
 
# File 'lib/geneva_drive/step_execution.rb', line 295

def logger
  @logger ||= begin
    workflow_tagged_logger = workflow.logger
    workflow_tagged_logger.tagged("execution_id=#{id} step_name=#{step_name}")
  end
end

#mark_canceled!(outcome: "canceled") ⇒ void

This method returns an undefined value.

Marks the step execution as canceled.

Parameters:

  • outcome (String) (defaults to: "canceled") —

    the outcome (defaults to 'canceled')



235
236
237
238
239
240
241
242
243
 
# File 'lib/geneva_drive/step_execution.rb', line 235

def mark_canceled!(outcome: "canceled")
  with_lock do
    update!(
      state: "canceled",
      canceled_at: Time.current,
      outcome: outcome
    )
  end
end

#mark_completed!(outcome: "success") ⇒ void

This method returns an undefined value.

Marks the step execution as completed.

Parameters:

  • outcome (String) (defaults to: "success") —

    the outcome ('success' or 'reattempted')



188
189
190
191
192
193
194
195
196
 
# File 'lib/geneva_drive/step_execution.rb', line 188

def mark_completed!(outcome: "success")
  with_lock do
    update!(
      state: "completed",
      completed_at: Time.current,
      outcome: outcome
    )
  end
end

#mark_failed!(error, outcome: "failed") ⇒ void

This method returns an undefined value.

Marks the step execution as failed and records the error.

Parameters:

  • error (Exception) —

    the error that occurred

  • outcome (String) (defaults to: "failed") —

    the outcome ('failed' or 'canceled')



203
204
205
206
207
208
209
210
211
212
213
214
215
 
# File 'lib/geneva_drive/step_execution.rb', line 203

def mark_failed!(error, outcome: "failed")
  with_lock do
    attrs = {
      state: "failed",
      failed_at: Time.current,
      outcome: outcome,
      error_message: error.message,
      error_backtrace: error.backtrace&.join("\n")
    }
    attrs[:error_class_name] = error.class.name if has_attribute?(:error_class_name)
    update!(attrs)
  end
end

#mark_skipped!(outcome: "skipped") ⇒ void

This method returns an undefined value.

Marks the step execution as skipped.

Parameters:

  • outcome (String) (defaults to: "skipped") —

    the outcome (defaults to 'skipped')



221
222
223
224
225
226
227
228
229
 
# File 'lib/geneva_drive/step_execution.rb', line 221

def mark_skipped!(outcome: "skipped")
  with_lock do
    update!(
      state: "skipped",
      skipped_at: Time.current,
      outcome: outcome
    )
  end
end

#resuming? ⇒ Boolean

Returns true if this execution is resuming from a prior execution. Always false when the continues_from_id column has not been migrated yet.

Returns:

  • (Boolean) 


272
273
274
 
# File 'lib/geneva_drive/step_execution.rb', line 272

def resuming?
  self.class.resumable_columns? && continues_from_id.present?
end

#start! ⇒ Boolean

Transitions the step execution to 'in_progress' state. Uses pessimistic locking to prevent double execution.

Returns:

  • (Boolean) —

    true if transition succeeded, false if already executed



173
174
175
176
177
178
179
180
181
182
 
# File 'lib/geneva_drive/step_execution.rb', line 173

def start!
  with_lock do
    return false unless scheduled?
    update!(
      state: "in_progress",
      started_at: Time.current
    )
  end
  true
end

#step_definition ⇒ StepDefinition?

Returns the step definition for this execution.

Returns:



279
280
281
 
# File 'lib/geneva_drive/step_execution.rb', line 279

def step_definition
  workflow.class.steps.named(step_name)
end

#with_logger(logger) { ... } ⇒ Object

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.

Temporarily overrides the step execution's logger for the duration of the block. The passed logger is used directly - it should already include step-specific tags. The original logger is restored after the block completes.

Parameters:

  • logger (Logger) —

    the logger to use (already fully tagged)

Yields:

  • the block to execute with the injected logger

Returns:

  • (Object) —

    the result of the block



310
311
312
313
314
315
316
 
# File 'lib/geneva_drive/step_execution.rb', line 310

def with_logger(logger)
  previous_logger = @logger
  @logger = logger
  yield
ensure
  @logger = previous_logger
end