Class: GenevaDrive::Workflow

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

Overview

Base class for all durable workflows in GenevaDrive.

Provides a DSL for defining multi-step workflows that execute asynchronously, with strong guarantees around idempotency, concurrency control, and state management.

Examples:

Basic workflow definition

class SignupWorkflow < GenevaDrive::Workflow
  step :send_welcome_email do
    WelcomeMailer.welcome(hero).deliver_later
  end

  step :send_reminder, wait: 2.days do
    ReminderMailer.remind(hero).deliver_later
  end
end

Creating and starting a workflow

SignupWorkflow.create!(hero: current_user)

Defined Under Namespace

Modules: MetadataAccessor

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from FlowControl

#cancel!, #finished!, #pause!, #reattempt!, #skip!, #suspend!

Methods included from MetadataAccessor

#read_metadata, #write_metadata

Instance Attribute Details

#received_signal ⇒ Object (readonly)

Returns the value of attribute received_signal.



1406
1407
1408
 
# File 'lib/geneva_drive/workflow.rb', line 1406

def received_signal
  @received_signal
end

Class Method Details

.cancel_if(*conditions) { ... } ⇒ void

This method returns an undefined value.

Defines a blanket cancellation condition for the workflow. Checked before every step execution.

Examples:

Cancel if hero is deactivated

cancel_if { hero.deactivated? }

Cancel using a method

cancel_if :hero_deactivated?

Parameters:

  • conditions (Array<Symbol, Proc>) —

    condition methods or procs

Yields:

  • an optional block condition



307
308
309
310
311
312
313
 
# File 'lib/geneva_drive/workflow.rb', line 307

def cancel_if(*conditions, &block)
  # Duplicate parent's array to avoid mutation
  self._cancel_conditions = _cancel_conditions.dup

  _cancel_conditions.concat(conditions)
  _cancel_conditions << block if block_given?
end

.may_proceed_without_hero! ⇒ void

This method returns an undefined value.

Allows the workflow to continue even if the hero is deleted. By default, workflows cancel if their hero is missing.

Examples:

Allow cleanup workflows to run without hero

class CleanupWorkflow < GenevaDrive::Workflow
  may_proceed_without_hero!

  step :cleanup do
    DataArchive.cleanup_for_hero_id(hero&.id)
  end
end


342
343
344
 
# File 'lib/geneva_drive/workflow.rb', line 342

def may_proceed_without_hero!
  self._may_proceed_without_hero = true
end

.on_exception(action, *exception_matchers, wait: nil, max_reattempts: nil, report: :always) ⇒ Object .on_exception(*exception_matchers, action:, wait: nil, max_reattempts: nil, report: :always) ⇒ Object .on_exception(*exception_matchers, report: :always) {|error| ... } ⇒ Object

Declares a class-level exception handling policy. Policies are checked when a step raises an exception and the step itself does not have an explicit on_exception: override.

Examples:

Blanket default for all exceptions

on_exception :reattempt!, wait: 15.seconds, max_reattempts: 3

Match specific exception classes

on_exception OAuth2::Error, action: :reattempt!, wait: 15.seconds
on_exception Google::Apis::ClientError, action: :cancel!

Imperative block handler

on_exception RateLimitError do |error|
  reattempt! wait: error.retry_after.seconds
end

Suppress reporting for expected rate limits

on_exception RateLimitError, report: :never do |error|
  reattempt! wait: error.retry_after.seconds
end

Report only when reattempts are exhausted

on_exception Timeout::Error, action: :reattempt!, max_reattempts: 5, report: :terminal_only

Overloads:

  • .on_exception(action, *exception_matchers, wait: nil, max_reattempts: nil, report: :always) ⇒ Object

    Declarative mode — specify an action symbol.

    Parameters:

    • action (Symbol) —

      :pause!, :cancel!, :reattempt!, or :skip!

    • exception_matchers (Array<Class>) —

      optional exception classes to match

    • wait (ActiveSupport::Duration, nil) (defaults to: nil) —

      wait before reattempt

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

      max consecutive reattempts

    • report (Symbol) (defaults to: :always) —

      when to report the exception to Rails.error.report (+:always+, :never, or :terminal_only). See ExceptionPolicy for details.

  • .on_exception(*exception_matchers, action:, wait: nil, max_reattempts: nil, report: :always) ⇒ Object

    Declarative mode with exception classes as leading args and action as keyword.

    Parameters:

    • exception_matchers (Array<Class>) —

      exception classes to match

    • action (Symbol) —

      :pause!, :cancel!, :reattempt!, or :skip!

    • report (Symbol) (defaults to: :always) —

      when to report the exception (+:always+, :never, or :terminal_only)

  • .on_exception(*exception_matchers, report: :always) {|error| ... } ⇒ Object

    Imperative mode — block receives exception, runs in workflow context.

    Parameters:

    • exception_matchers (Array<Class>) —

      optional exception classes to match

    • report (Symbol) (defaults to: :always) —

      when to report the exception (+:always+, :never, or :terminal_only)

    Yields:

    • (error) —

      the exception that was raised



390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
 
# File 'lib/geneva_drive/workflow.rb', line 390

def on_exception(*args, action: nil, wait: nil, max_reattempts: nil, terminal_action: :pause!, report: :always, &block)
  # Separate exception classes from a leading action symbol
  if args.first.is_a?(Symbol)
    raise ArgumentError, "Cannot pass both a positional action and action: keyword" if action
    action = args.shift
  end

  exception_matchers = args.map do |matcher|
    if matcher.is_a?(String)
      GenevaDrive::ExceptionPolicy::LazyExceptionMatcher.new(matcher)
    elsif matcher.is_a?(Class)
      unless matcher <= Exception
        raise GenevaDrive::StepConfigurationError,
          "Expected an Exception subclass, got #{matcher.inspect}"
      end
      matcher
    elsif matcher.respond_to?(:===)
      matcher
    else
      raise GenevaDrive::StepConfigurationError,
        "Expected an exception matcher (Exception subclass, String, or object responding to #===), got #{matcher.inspect}"
    end
  end

  policy = if block
    GenevaDrive::ExceptionPolicy.new(report: report, &block)
  else
    raise ArgumentError, "Either an action or a block is required" unless action
    GenevaDrive::ExceptionPolicy.new(action, wait: wait, max_reattempts: max_reattempts, terminal_action: terminal_action, report: report)
  end

  policy.exception_matchers.concat(exception_matchers)

  # Duplicate parent's array to avoid mutation
  if _exception_policies.equal?(superclass._exception_policies)
    self._exception_policies = _exception_policies.dup
  end

  _exception_policies << policy
end

.removed_step(name) ⇒ RemovedStepDefinition

Marks a step as removed, leaving a gravestone in the position the step used to occupy.

Deleting a step outright is unsafe while workflows are in flight. Executions scheduled under the old name find no definition when their job runs, and the workflow pauses with StepNotDefinedError - stuck until an operator intervenes. The gem cannot recover from this on its own: it knows the step is gone, but not whether the work it did still needs doing, and not which of the remaining steps would have run had it executed. Only the author of the removal knows that.

removed_step is how that knowledge gets written down. Declaring it where the step used to be says "this step no longer needs to happen, carry on past it", and the gem can then act without guessing:

  • Executions already scheduled for the name resolve, are skipped, and the workflow proceeds to the next runnable step.
  • Scheduling never spools a new execution for the name again.
  • before_step: / after_step: references to it keep working.

Ship the removed_step, let in-flight workflows drain past it, then delete the line once removed_steps_in_flight comes back empty.

A removed step takes a name and nothing else. There is nothing to configure about a step that does not run.

Examples:

Taking a step out of a live workflow

class PaymentWorkflow < GenevaDrive::Workflow
  step :authorize do
    # ...
  end

  removed_step :capture_payment

  step :send_receipt do
    # ...
  end
end

Parameters:

  • name (String, Symbol) —

    the name of the step that was removed

Returns:

Raises:

  • (ArgumentError) 


250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
 
# File 'lib/geneva_drive/workflow.rb', line 250

def removed_step(name)
  raise ArgumentError, "removed_step requires a step name" if name.nil?

  caller_loc = caller_locations(1, 1).first
  call_location = caller_loc ? [caller_loc.path, caller_loc.lineno] : nil

  step_name = prepare_step_registration!(name, {})

  step_def = GenevaDrive::RemovedStepDefinition.new(
    name: step_name,
    call_location: call_location
  )

  _step_definitions << step_def

  step_def
end

.removed_steps_in_flight ⇒ Hash{String => Integer}

Returns the names of this workflow's removed steps that step execution rows still reference, so a removed_step gravestone is only deleted once nothing can land on it any more.

Each name maps to the number of rows still referencing it. An empty hash means every in-flight workflow has drained past the removal and the removed_step lines can go.

Rows belonging to finished and canceled workflows are ignored - they will never execute again.

Examples:

PaymentWorkflow.removed_steps_in_flight
# => {"capture_payment" => 3}

Returns:

  • (Hash{String => Integer}) —

    removed step names to referencing row counts



284
285
286
287
288
289
290
291
292
293
 
# File 'lib/geneva_drive/workflow.rb', line 284

def removed_steps_in_flight
  removed_names = steps.removed.map(&:name)
  return {} if removed_names.empty?

  GenevaDrive::StepExecution
    .where(step_name: removed_names)
    .where(workflow_id: where.not(state: %w[finished canceled]).select(:id))
    .group(:step_name)
    .count
end

.resolve_exception_policy(error) ⇒ ExceptionPolicy?

Resolves the class-level exception policy for a given error. Checks policies in reverse definition order (most recent first). Specific (exception class) policies are checked before blanket policies.

Parameters:

  • error (Exception) —

    the exception to match

Returns:



437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
 
# File 'lib/geneva_drive/workflow.rb', line 437

def resolve_exception_policy(error)
  # Walk in reverse order (most recently defined first).
  # Check specific policies (with exception class filters) first,
  # then blanket policies (no filters).
  blanket_policy = nil

  _exception_policies.reverse_each do |policy|
    if policy.specific?
      return policy if policy.matches?(error)
    else
      # Remember the first (most recent) blanket policy
      blanket_policy ||= policy
    end
  end

  blanket_policy
end

.resumable_step(name = nil, **options) {|iter| ... } ⇒ void

This method returns an undefined value.

Defines a resumable step that can iterate over large collections. The block receives an IterableStep object for cursor-based iteration that survives job restarts.

Examples:

Iterate over records with automatic checkpointing

resumable_step :process_users do |iter|
  iter.iterate_over_records(hero.users) do |user|
    process(user)
  end
end

Manual cursor control

resumable_step :sync_pages do |iter|
  page = iter.cursor || 1
  loop do
    response = Api.fetch(page: page)
    break if response.empty?
    response.each { |item| process(item) }
    page += 1
    iter.set!(page)
  end
end

With iteration limits

resumable_step :bulk_import, max_iterations: 10_000 do |iter|
  iter.iterate_over_records(records) { |r| import(r) }
end

Parameters:

  • name (String, Symbol, nil) (defaults to: nil) —

    the step name (auto-generated if nil)

  • options (Hash) —

    step options

Options Hash (**options):

  • :max_iterations (Integer, nil) —

    interrupt after N iterations

  • :max_runtime (ActiveSupport::Duration, nil) —

    interrupt after duration

  • :wait (ActiveSupport::Duration, nil) —

    delay before execution

  • :skip_if (Proc, Symbol, Boolean, nil) —

    condition for skipping

  • :on_exception (Symbol) —

    exception handler (:pause!, :cancel!, :reattempt!, :skip!)

  • :before_step (String, Symbol, nil) —

    position before this step

  • :after_step (String, Symbol, nil) —

    position after this step

Yields:

  • (iter) —

    the step implementation receiving an IterableStep object

Yield Parameters:

Raises:

  • (ArgumentError) 


186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
 
# File 'lib/geneva_drive/workflow.rb', line 186

def resumable_step(name = nil, **options, &block)
  raise ArgumentError, "resumable_step requires a block" unless block_given?

  # Capture source locations before any other operations
  caller_loc = caller_locations(1, 1).first
  call_location = caller_loc ? [caller_loc.path, caller_loc.lineno] : nil
  block_location = block.source_location

  step_name = prepare_step_registration!(name, options)

  step_def = GenevaDrive::ResumableStepDefinition.new(
    name: step_name,
    call_location: call_location,
    block_location: block_location,
    **options,
    &block
  )

  _step_definitions << step_def

  step_def
end

.set_step_job_options(**options) ⇒ void

This method returns an undefined value.

Sets job options for step execution jobs. Options are passed to ActiveJob's set method.

Examples:

Set queue for workflow steps

set_step_job_options queue: :workflows, priority: 10

Parameters:

  • options (Hash) —

    job options (queue, priority, etc.)



323
324
325
326
327
 
# File 'lib/geneva_drive/workflow.rb', line 323

def set_step_job_options(**options)
  validated = GenevaDrive::JobOptions.validate!(options, context: "#{name || "Workflow"}.set_step_job_options")
  # Merge with parent's options
  self._step_job_options = _step_job_options.merge(validated)
end

.step(name = nil, **options) { ... } ⇒ GenevaDrive::StepDefinition

Defines a step in the workflow.

Examples:

Named step with block

step :send_email do
  Mailer.send(hero).deliver_later
end

Step with wait time

step :send_reminder, wait: 2.days do
  ReminderMailer.remind(hero).deliver_later
end

Step with skip condition

step :charge, skip_if: -> { hero.free_tier? } do
  PaymentGateway.charge(hero)
end

Step with simple exception handling

step :external_api, on_exception: :reattempt! do
  ExternalApi.call(hero)
end

Step with composable exception policies

step :sync_calendar, on_exception: [
  GenevaDrive::ExceptionPolicy.new(:reattempt!, matching: Timeout::Error, max_reattempts: 5),
  GenevaDrive::ExceptionPolicy.new(:cancel!, matching: OAuth2::Error),
  GenevaDrive::ExceptionPolicy.new(:skip!)  # blanket fallback
] do
  GoogleCalendar.sync(hero)
end

Parameters:

  • name (String, Symbol, nil) (defaults to: nil) —

    the step name (auto-generated if nil)

  • options (Hash) —

    step options

Options Hash (**options):

  • :wait (ActiveSupport::Duration, nil) —

    delay before execution

  • :job_options (Hash) —

    options passed to Active Job's set method

  • :skip_if (Proc, Symbol, Boolean, nil) —

    condition for skipping

  • :on_exception (Symbol, GenevaDrive::ExceptionPolicy, Proc, Array<GenevaDrive::ExceptionPolicy>) —

    exception handling policy. Accepts:

    • A Symbol (+:pause!+, :cancel!, :reattempt!, :skip!) for simple actions
    • An ExceptionPolicy object for reusable, configurable policies
    • A Proc/lambda that receives the exception and calls a flow control method
    • An Array of ExceptionPolicy objects for composable per-exception-type handling. Specific policies (those with matching:) are checked first; the first blanket policy acts as a fallback. If nothing matches, class-level policies are consulted.
  • :max_reattempts (Integer, nil) —

    max consecutive reattempts (only with symbol form)

  • :terminal_action (Symbol) —

    what to do when max_reattempts is exceeded (+:pause!+ or :cancel!, only with symbol form)

  • :before_step (String, Symbol, nil) —

    position before this step

  • :after_step (String, Symbol, nil) —

    position after this step

Yields:

  • the step implementation

Returns:



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
 
# File 'lib/geneva_drive/workflow.rb', line 125

def step(name = nil, **options, &block)
  # Capture source locations before any other operations
  caller_loc = caller_locations(1, 1).first
  call_location = caller_loc ? [caller_loc.path, caller_loc.lineno] : nil
  block_location = block&.source_location

  step_name = prepare_step_registration!(name, options)

  step_def = GenevaDrive::StepDefinition.new(
    name: step_name,
    callable: block || name,
    call_location: call_location,
    block_location: block_location,
    **options
  )

  _step_definitions << step_def

  step_def
end

.step_definitions ⇒ Array<StepDefinition>

Returns the step definitions for this workflow class.

Returns:



458
459
460
 
# File 'lib/geneva_drive/workflow.rb', line 458

def step_definitions
  _step_definitions
end

.steps ⇒ StepCollection

Returns the step collection with proper ordering.

Returns:



465
466
467
 
# File 'lib/geneva_drive/workflow.rb', line 465

def steps
  @steps ||= GenevaDrive::StepCollection.new(_step_definitions)
end

Instance Method Details

#after_step_execution(step_execution) ⇒ void

This method returns an undefined value.

Hook called after step code completes, before finalization. Called regardless of whether the step succeeded, failed, or used flow control.

Parameters:



981
982
983
 
# File 'lib/geneva_drive/workflow.rb', line 981

def after_step_execution(step_execution)
  # Override in subclasses
end

#around_step_execution(step_execution) { ... } ⇒ Object

Hook that wraps around the actual step code execution. Use this for APM instrumentation that requires wrapping a block.

IMPORTANT: Subclasses MUST call super when overriding this method, otherwise the step code will not execute.

Examples:

Wrap with AppSignal transaction

def around_step_execution(step_execution)
  Appsignal.monitor(
    namespace: "workflow",
    action: "#{self.class.name}##{step_execution.step_name}"
  ) { super }
end

Parameters:

Yields:

  • the step code block

Returns:

  • (Object) —

    the result of the block



1002
1003
1004
 
# File 'lib/geneva_drive/workflow.rb', line 1002

def around_step_execution(step_execution)
  yield
end

#attachable_signals ⇒ Array<GenevaDrive::Signal>

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.

Signals that may still be attached to an execution, oldest first, each pointing back at this workflow instance.

Matcher blocks are instance_exec'd on signal.workflow, and inverse_of: only primes records loaded straight off the association - a scoped relation would hand each signal a freshly loaded workflow, which would then see none of the caller's unsaved state.

Returns:



840
841
842
 
# File 'lib/geneva_drive/workflow.rb', line 840

public def attachable_signals
  signals.attachable.map { |signal| adopt_signal(signal) }
end

#before_step_execution(step_execution) ⇒ void

This method returns an undefined value.

Hook called before step execution, after validation passes. Use this for APM instrumentation like setting AppSignal action/params.

Examples:

Set AppSignal transaction metadata

def before_step_execution(step_execution)
  Appsignal.set_action("#{self.class.name}##{step_execution.step_name}")
  Appsignal.set_params("hero" => { "type" => hero_type, "id" => hero_id })
end

Parameters:



972
973
974
 
# File 'lib/geneva_drive/workflow.rb', line 972

def before_step_execution(step_execution)
  # Override in subclasses
end

#create_successor_execution!(predecessor, wait: nil) ⇒ StepExecution

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.

Creates a successor step execution that continues a resumable step from the cursor persisted on the given execution. Used by the Executor when a resumable step interrupts itself, by resume! for workflows paused mid-iteration, and by housekeeping recovery of stuck resumable steps.

The successor is enqueued with the same merged job options (class, per-instance, per-step) as any other execution of the step.

Parameters:

  • predecessor (StepExecution) —

    the execution to continue from

  • wait (ActiveSupport::Duration, Numeric, nil) (defaults to: nil) —

    optional delay

Returns:



1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
 
# File 'lib/geneva_drive/workflow.rb', line 1216

public def create_successor_execution!(predecessor, wait: nil)
  scheduled_for = wait ? wait.from_now : Time.current

  with_lock do
    # Cancel any stray scheduled or parked executions - the successor is the
    # one execution that should run next (same as create_step_execution).
    cancel_stray_executions!

    successor_attributes = {
      step_name: predecessor.step_name,
      state: "scheduled",
      scheduled_for: scheduled_for,
      continues_from_id: predecessor.id,
      cursor: predecessor.cursor
    }
    # Carry the attachment pin across the chain so successors never re-park
    # and received_signal stays stable for the whole iteration.
    if GenevaDrive::StepExecution.signal_columns?
      successor_attributes[:signal_id] = predecessor.signal_id
    end

    successor = step_executions.create!(**successor_attributes)

    # next_step_name points to the step that's scheduled to run next
    update!(next_step_name: predecessor.step_name)

    job_options = merged_step_job_options(predecessor.step_definition)
    job_options[:wait_until] = scheduled_for if wait
    successor_id = successor.id
    workflow_logger = logger

    run_after_commit do
      job = GenevaDrive::PerformStepJob
        .set(**job_options)
        .perform_later(successor_id)

      workflow_logger.debug("Enqueued PerformStepJob with job_id=#{job.job_id} for successor execution ##{successor_id}")

      GenevaDrive::StepExecution
        .where(id: successor_id)
        .update_all(job_id: job.job_id)
    end

    successor
  end
end

#current_execution ⇒ StepExecution?

Returns the current active step execution, if any. Includes scheduled, waiting and in_progress states.

Returns:



826
827
828
 
# File 'lib/geneva_drive/workflow.rb', line 826

def current_execution
  step_executions.where(state: %w[scheduled waiting in_progress]).first
end

#dispatch_signal!(signal) ⇒ Array<StepExecution>

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.

Wakes every parked step execution whose matcher accepts the given signal.

Runs under the workflow row lock - the same lock create_step_execution and the Executor take - so whichever of dispatch and the gate commits second sees the other's write and no signal can slip through the gap.

Parameters:

Returns:



853
854
855
856
857
 
# File 'lib/geneva_drive/workflow.rb', line 853

public def dispatch_signal!(signal)
  return [] unless GenevaDrive::StepExecution.signal_columns?

  with_lock { wake_executions_matching!(signal) }
end

#execution_history ⇒ ActiveRecord::Relation<StepExecution>

Returns all step executions in chronological order.

Returns:

  • (ActiveRecord::Relation<StepExecution>) —

    the execution history



930
931
932
 
# File 'lib/geneva_drive/workflow.rb', line 930

def execution_history
  step_executions.order(:created_at)
end

#logger ⇒ Logger

Returns the Logger properly tagged to this Workflow

Returns:

  • (Logger) 


1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
 
# File 'lib/geneva_drive/workflow.rb', line 1427

public def logger
  @tagged_logger ||= begin
    base_logger = @injected_base_logger || super
    tagged_logger = ActiveSupport::TaggedLogging.new(base_logger)

    # Tag log entries with the workflow, including hero info if present.
    # Step name is logged separately via the StepExecution logger.
    tag_parts = [self.class.name, " id=", to_param]
    if hero_id.present?
      tag_parts.concat([" hero_type=", hero_type, " hero_id=", hero_id])
    end
    workflow_tag = tag_parts.join

    tagged_logger.tagged(workflow_tag)
  end
end

#ongoing? ⇒ Boolean

Returns whether this workflow is in an ongoing (non-terminal) state.

Returns:

  • (Boolean) 


541
542
543
 
# File 'lib/geneva_drive/workflow.rb', line 541

def ongoing?
  !finished? && !canceled?
end

#previous_step_name ⇒ String?

Returns the name of the previously executed step.

Logic:

  • If current_step_name is set (currently executing), returns the step before it
  • If only next_step_name is set (waiting for next step), returns the step before it
  • If workflow is finished (no next step), returns the last step in the sequence
  • Returns nil if this is the first step or no steps have been executed

Returns:

  • (String, nil) —

    the previous step name or nil



950
951
952
953
954
955
956
957
958
959
 
# File 'lib/geneva_drive/workflow.rb', line 950

def previous_step_name
  reference_step = current_step_name || next_step_name

  if reference_step
    previous_step = steps.previous_before(reference_step)
    previous_step&.name
  elsif finished?
    steps.last&.name
  end
end

#reschedule_current_step!(wait: nil) ⇒ StepExecution

Reschedules the current step for another attempt.

Uses current_step_name if executing, otherwise next_step_name.

Parameters:

  • wait (ActiveSupport::Duration, nil) (defaults to: nil) —

    delay before retry

Returns:



570
571
572
573
574
575
576
577
 
# File 'lib/geneva_drive/workflow.rb', line 570

def reschedule_current_step!(wait: nil)
  # Use current_step_name during execution, next_step_name otherwise
  step_name = current_step_name || next_step_name
  step_def = steps.named(step_name)
  wait_msg = wait ? " with wait #{wait.inspect}" : ""
  logger.info("Rescheduling step #{step_name.inspect}#{wait_msg}")
  create_step_execution(step_def, wait: wait)
end

#resume! ⇒ StepExecution?

Resumes a paused workflow.

Scheduling behavior

Since pause! leaves the scheduled execution intact, resume! re-enqueues a job for the existing execution:

  • Scheduled time still in future: Enqueues job with remaining wait time
  • Scheduled time has passed (overdue): Enqueues job to run immediately
  • No scheduled execution exists: Creates a new execution for immediate run (This happens if the executor ran while paused and canceled the execution)

This approach provides better timeline visibility - you can see that a step was scheduled, became overdue during pause, and when it actually ran.

Examples:

Resuming while step is still scheduled for future

# step_two has wait: 2.days, scheduled for tomorrow
workflow.pause!         # paused today
workflow.resume!        # step_two still scheduled for tomorrow

Resuming after scheduled time passed (overdue)

# step_two was scheduled for yesterday
workflow.pause!         # paused last week
workflow.resume!        # step_two runs immediately (overdue)

Returns:

  • (StepExecution, nil) —

    the step execution that will run, or nil if none

Raises:



606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
 
# File 'lib/geneva_drive/workflow.rb', line 606

def resume!
  raise GenevaDrive::InvalidStateError, "Cannot resume a #{state} workflow" unless state == "paused"

  logger.info("Resuming paused workflow, next step: #{next_step_name.inspect}")

  with_lock do
    update!(state: "ready", transitioned_at: nil)
  end

  # A scheduled execution preserved by pause! takes precedence - re-enqueue it
  scheduled_execution = current_execution
  if scheduled_execution
    # A parked execution is not re-enqueued: it is waiting for an event, not
    # for a clock. Re-run the rendezvous instead, which delivers any signal
    # that arrived while the workflow was paused.
    return rendezvous_waiting_execution!(scheduled_execution) if scheduled_execution.waiting?

    return enqueue_scheduled_execution(scheduled_execution)
  end

  # Resumable-step continuations need the cursor/continues_from_id columns
  if GenevaDrive::StepExecution.resumable_columns?
    # Check for a resumable step that was paused mid-iteration. Only the most
    # recent marker row counts, and only while it has not been consumed yet
    # (creating a successor consumes it - the successor's presence marks that).
    paused_resumable_execution = step_executions
      .where(outcome: "workflow_paused", state: "completed")
      .order(created_at: :desc, id: :desc)
      .first

    if paused_resumable_execution && paused_resumable_execution.successor.nil?
      logger.info("Resuming from resumable execution #{paused_resumable_execution.id}, cursor: #{paused_resumable_execution.cursor_value.inspect}")
      return create_successor_execution!(paused_resumable_execution)
    end

    # A resumable step that failed (pausing the workflow) resumes from its
    # persisted cursor instead of redoing the whole iteration.
    failed_execution = step_executions
      .where(step_name: next_step_name, state: "failed")
      .order(created_at: :desc, id: :desc)
      .first
    if failed_execution && failed_execution.cursor.present? && failed_execution.successor.nil?
      logger.info("Retrying failed resumable execution #{failed_execution.id} from cursor: #{failed_execution.cursor_value.inspect}")
      return create_successor_execution!(failed_execution)
    end
  end

  # No scheduled execution exists - create one for the next step
  step_def = steps.named(next_step_name)

  # The step this workflow is pointed at can be gone from the code - that is
  # what paused it in the first place if it paused with StepNotDefinedError.
  # Say so plainly instead of dying on nil deeper in create_step_execution,
  # and name the way out.
  unless step_def
    raise GenevaDrive::StepNotDefinedError.new(
      "Cannot resume #{self.class.name} ##{id}: step '#{next_step_name}' is no longer defined in " \
      "#{self.class.name}. Pick the step to continue from with resume_at!(:step_name) - the steps " \
      "still defined are #{steps.runnable.map(&:name).join(", ")} - or cancel! the workflow. To stop " \
      "future removals from stranding workflows this way, leave a removed_step :#{next_step_name} " \
      "gravestone in place of the deleted step.",
      step_execution: nil,
      workflow: self
    )
  end

  create_step_execution(step_def, wait: nil)
end

#resume_at!(step_name) ⇒ StepExecution

Resumes a paused workflow at a specific step, instead of at whatever step it was pointed at when it paused.

This is the operator's escape hatch for a workflow stranded on a step that no longer exists - the step was deleted without leaving a removed_step gravestone, so the workflow paused with StepNotDefinedError and plain #resume! has nowhere to go.

Where to land is deliberately a human decision. A deleted step's position in the sequence says nothing about whether the steps that follow it were ever meant to run: a capture_payment that normally ends the workflow by calling finished! sits directly before refund, and nothing in the database distinguishes "carry on" from "stop here". Only the person who removed the step knows which, so this method asks them.

Any stray scheduled execution is canceled first, so the workflow continues from the named step and nowhere else.

Examples:

Continue a stranded workflow past a deleted step

workflow = GenevaDrive::Workflow.find(id)
workflow.resume_at!(:send_receipt)

Parameters:

  • step_name (String, Symbol) —

    the step to continue from

Returns:

Raises:



701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
 
# File 'lib/geneva_drive/workflow.rb', line 701

def resume_at!(step_name)
  raise GenevaDrive::InvalidStateError, "Cannot resume a #{state} workflow" unless state == "paused"

  step_def = steps.named(step_name.to_s)

  if step_def.nil? || step_def.removed?
    reason = step_def ? "a removed_step" : "not defined"
    raise GenevaDrive::StepNotDefinedError.new(
      "Cannot resume #{self.class.name} ##{id} at '#{step_name}': that step is #{reason} in " \
      "#{self.class.name}. The steps it can be resumed at are #{steps.runnable.map(&:name).join(", ")}.",
      step_execution: nil,
      workflow: self
    )
  end

  logger.info("Resuming paused workflow at step #{step_def.name.inspect} (was pointed at #{next_step_name.inspect})")

  with_lock do
    update!(state: "ready", current_step_name: nil, transitioned_at: nil)
  end

  create_step_execution(step_def, wait: nil)
end

#schedule_next_step!(wait: nil) ⇒ StepExecution?

Schedules the next step in the workflow.

Uses current_step_name as reference if executing, otherwise next_step_name.

Parameters:

  • wait (ActiveSupport::Duration, nil) (defaults to: nil) —

    override wait time

Returns:

  • (StepExecution, nil) —

    the created step execution or nil if finished



551
552
553
554
555
556
557
558
559
560
561
562
 
# File 'lib/geneva_drive/workflow.rb', line 551

def schedule_next_step!(wait: nil)
  # Use current_step_name during execution, next_step_name otherwise
  reference_step = current_step_name || next_step_name
  next_step = steps.next_after(reference_step)
  unless next_step
    logger.info("No more steps after #{reference_step.inspect}, finishing workflow")
    return finish_workflow!
  end

  logger.info("Scheduling next step #{next_step.name.inspect} after #{reference_step.inspect}")
  create_step_execution(next_step, wait: wait || next_step.wait)
end

#settle_signal_for_skipped!(step_execution) ⇒ GenevaDrive::Signal?

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.

Settles the signal attached to an execution we are deliberately moving past, wherever the skip came from: skip_if firing at wake, flow control inside the step, an exception policy, or an operator calling skip!.

Consumption follows attachment. A skip in any flavor says "we are done with this step", so an event that was delivered to it is spent by that decision - leaving it claimed would hand a stale event to the next waiter for the same name, which is exactly what consume-once-per-event forbids. An execution that never attached has nothing to settle, so buffered signals a step never reached are left alone.

Must be called with the workflow lock held, so the settle rides the same transaction as the skip itself.

Parameters:

  • step_execution (StepExecution, nil) —

    the execution being skipped past

Returns:



876
877
878
879
880
881
882
883
884
885
886
 
# File 'lib/geneva_drive/workflow.rb', line 876

public def settle_signal_for_skipped!(step_execution)
  return unless GenevaDrive::StepExecution.signal_columns?
  return if step_execution.nil? || step_execution.signal_id.blank?

  signal = step_execution.signal
  return unless signal && signal.state == "claimed"

  logger.info("Skipping past step #{step_execution.step_name}, settling attached signal #{signal.id} (#{signal.name})")
  signal.record_consumption!(resolved_step_names: [step_execution.step_name])
  signal
end

#signal!(signal_name, payload: {}, idempotency_key: nil, **unknown_options) ⇒ GenevaDrive::Signal

Delivers an external event to this workflow.

The signal row is persisted first and dispatched second, which is what makes arrival order irrelevant: a signal that lands before the waiting step's execution even exists simply sits in the table until that step's gate picks it up, and a signal that lands while a step is already parked wakes it immediately.

Pass an idempotency_key (a webhook's event id, typically) to make redelivery a no-op: the second call returns the row the first one wrote, flagged Signal#duplicate_delivery?, without dispatching.

On a paused workflow the row is persisted but not dispatched - waking a step while the workflow is paused would only get its execution canceled. #resume! performs the rendezvous instead.

Examples:

Deliver a webhook event

workflow = OrderFulfillmentWorkflow.ongoing.for_hero(order).first
workflow.signal!(:payment_confirmed,
  payload: {amount_cents: 12_500},
  idempotency_key: event["id"])

Parameters:

  • signal_name (Symbol, String) —

    the event name

  • payload (Object) (defaults to: {}) —

    data describing the event, serialized through ActiveJob

  • idempotency_key (String, nil) (defaults to: nil) —

    deduplication key, scoped to (workflow, name)

Returns:

Raises:

  • (ArgumentError) —

    if the signal name is blank

  • (WorkflowNotOngoing) —

    if the workflow is finished or canceled and this is a new event

  • (SignalPayloadTooLargeError) —

    if the payload exceeds GenevaDrive.max_signal_payload_size



754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
 
# File 'lib/geneva_drive/workflow.rb', line 754

def signal!(signal_name, payload: {}, idempotency_key: nil, **unknown_options)
  if unknown_options.any?
    raise ArgumentError,
      "Unknown options passed to signal!: #{unknown_options.keys.join(", ")}"
  end

  if signal_name.blank?
    raise ArgumentError, "signal! requires a signal name"
  end

  name = signal_name.to_s
  dedup_key = idempotency_key&.to_s

  # Serialize (and bound) before anything is persisted.
  serialized_payload = GenevaDrive::Signal.serialize_payload(payload)

  signal = nil
  duplicate = false

  # One transaction covers the lot: the INSERT, the waiting-execution scan,
  # and every attach, flip and counter bump. Taking the lock before writing
  # anything is what makes delivery all-or-nothing - insert first and
  # dispatch second, and a crash in between would leave a pending signal
  # sitting next to a waiting execution with nothing left to introduce them.
  with_lock do
    # with_lock reloads; the terminal check belongs inside it
    unless ongoing?
      existing = dedup_key && signals.find_by(name: name, idempotency_key: dedup_key)
      unless existing
        raise GenevaDrive::WorkflowNotOngoing,
          "Cannot deliver signal #{name.inspect} to a #{state} workflow"
      end

      logger.info("Signal #{name.inspect} redelivered to #{state} workflow, returning existing row")
      signal = existing
      duplicate = true
      next
    end

    begin
      # A savepoint so that hitting the dedup index does not poison the
      # transaction - the caller's, if they had one open, or ours.
      transaction(requires_new: true) do
        record = signals.new(name: name, idempotency_key: dedup_key, state: "pending")
        record[:payload] = serialized_payload
        record.save!
        signal = record
      end
    rescue ActiveRecord::RecordNotUnique
      signal = signals.find_by!(name: name, idempotency_key: dedup_key)
      logger.info("Signal #{name.inspect} is a duplicate delivery of signal #{signal.id}, not dispatching")
      duplicate = true
      next
    end

    logger.info("Received signal #{name.inspect} as signal #{signal.id}")

    if paused?
      logger.info("Workflow is paused, buffering signal #{signal.id} until resume!")
    else
      wake_executions_matching!(signal)
    end
  end

  signal.duplicate_delivery! if duplicate
  signal
end

#step_job_options ⇒ Hash

Returns per-instance step job options stored in metadata. These are merged with the class-level options when enqueuing step jobs, with instance options taking precedence.

Returns:

  • (Hash) —

    symbolized job options (e.g. { queue: :high, priority: 5 })



1011
1012
1013
 
# File 'lib/geneva_drive/workflow.rb', line 1011

def step_job_options
  (read_metadata("step_job_options") || {}).symbolize_keys
end

#step_job_options=(options) ⇒ void

This method returns an undefined value.

Sets per-instance step job options in metadata. Stored as part of the workflow's metadata column so they persist and survive across step boundaries.

Parameters:

  • options (Hash, nil) —

    job options to store



1021
1022
1023
1024
1025
1026
1027
1028
 
# File 'lib/geneva_drive/workflow.rb', line 1021

def step_job_options=(options)
  if options.present?
    validated = GenevaDrive::JobOptions.validate!(options, context: "#{self.class.name}#step_job_options=")
    write_metadata("step_job_options", validated.stringify_keys)
  else
    write_metadata("step_job_options", nil)
  end
end

#steps ⇒ StepCollection

Returns the step collection for this workflow's class.

Returns:



937
938
939
 
# File 'lib/geneva_drive/workflow.rb', line 937

def steps
  self.class.steps
end

#transition_to!(new_state, **attributes) ⇒ void

This method returns an undefined value.

Transitions the workflow to a new state.

Parameters:

  • new_state (String) —

    the target state

  • attributes (Hash) —

    additional attributes to update



1035
1036
1037
1038
1039
1040
1041
1042
1043
 
# File 'lib/geneva_drive/workflow.rb', line 1035

def transition_to!(new_state, **attributes)
  with_lock do
    attrs = attributes.merge(state: new_state)
    if %w[finished canceled paused].include?(new_state)
      attrs[:transitioned_at] = Time.current
    end
    update!(attrs)
  end
end

#wake_executions_matching!(signal) ⇒ Array<StepExecution>

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.

The body of dispatch, for callers that already hold the workflow lock (signal! holds it across the whole delivery so the rows land atomically).

Parameters:

Returns:



894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
 
# File 'lib/geneva_drive/workflow.rb', line 894

public def wake_executions_matching!(signal)
  return [] unless GenevaDrive::StepExecution.signal_columns?
  return [] if paused? || !ongoing?

  # Matcher blocks run on signal.workflow, so point the signal at this very
  # instance instead of letting it load a second copy from the database.
  adopt_signal(signal)
  woken = []

  step_executions.where(state: "waiting").order(created_at: :asc, id: :asc).each do |execution|
    step_def = execution.step_definition
    # A step whose definition is gone (class removed, step renamed) can
    # never match - the matcher lives on the definition.
    next unless step_def&.waits_for_signal?
    next unless step_def.signal_matcher.matches?(signal)

    signal.claim!
    execution.update!(
      signal_id: signal.id,
      state: "scheduled",
      scheduled_for: Time.current,
      waiting_since: nil
    )
    woken << [execution, step_def]
  end

  woken.each do |execution, step_def|
    enqueue_woken_execution(execution, step_def)
  end

  woken.map(&:first)
end

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

Temporarily overrides the workflow's logger for the duration of the block. This allows the Executor to inject the step execution's logger (which includes both workflow and step-specific tags) so that step code calling logger gets the fully-tagged step execution logger.

The passed logger is used directly - no additional tagging is applied. The original logger is restored after the block completes.

Examples:

Use step execution logger during step code

workflow.with_logger(step_execution.logger) do
  step_def.execute_in_context(workflow)
end

Parameters:

  • logger (Logger) —

    the logger to use (typically the step execution's logger)

Yields:

  • the block to execute with the injected logger

Returns:

  • (Object) —

    the result of the block



1389
1390
1391
1392
1393
1394
1395
 
# File 'lib/geneva_drive/workflow.rb', line 1389

public def with_logger(logger)
  previous_tagged_logger = @tagged_logger
  @tagged_logger = logger
  yield
ensure
  @tagged_logger = previous_tagged_logger
end

#with_received_signal(signal) { ... } ⇒ 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 makes a signal available to step code as received_signal for the duration of the block. Injected by the Executor the same way the tagged logger is.

Parameters:

Yields:

  • the block to execute with the signal in scope

Returns:

  • (Object) —

    the result of the block



1416
1417
1418
1419
1420
1421
1422
 
# File 'lib/geneva_drive/workflow.rb', line 1416

public def with_received_signal(signal)
  previous_received_signal = @received_signal
  @received_signal = signal
  yield
ensure
  @received_signal = previous_received_signal
end