Module: GenevaDrive::FlowControl

Included in:
Workflow
Defined in:
lib/geneva_drive/flow_control.rb

Overview

Module providing flow control methods for use within workflow steps. These methods use throw/catch to interrupt step execution and signal the executor how to proceed.

When called outside of a step execution context (e.g., from a controller), pause! and skip! will directly modify the workflow state.

Examples:

Using flow control in a step

step :process_payment do
  result = PaymentGateway.charge(hero)
  cancel! if result.declined?
  skip! if result.already_processed?
end

External flow control

workflow = MyWorkflow.find(id)
workflow.pause!  # Pauses workflow from outside a step

Instance Method Summary collapse

Instance Method Details

#cancel! ⇒ void

This method returns an undefined value.

Cancels the workflow immediately.

When called inside a step (workflow state is 'performing'): interrupts execution via throw/catch. When called outside a step (workflow state is 'ready' or 'paused'): directly cancels the workflow.

Raises:

  • (InvalidStateError) —

    if called on a non-ready/non-paused/non-performing workflow



119
120
121
122
123
124
125
126
 
# File 'lib/geneva_drive/flow_control.rb', line 119

def cancel!
  if state == "performing"
    logger.info("Flow control: cancel! called from step")
    throw :flow_control, GenevaDrive::FlowControlSignal.new(:cancel)
  else
    external_cancel!
  end
end

#finished! ⇒ void

This method returns an undefined value.

Marks the workflow as finished immediately. Useful for early termination when no further steps are needed.

Raises:

  • (UncaughtThrowError) —

    if called outside of step execution context



190
191
192
193
 
# File 'lib/geneva_drive/flow_control.rb', line 190

def finished!
  logger.info("Flow control: finished! called from step")
  throw :flow_control, GenevaDrive::FlowControlSignal.new(:finished)
end

#pause! ⇒ void

This method returns an undefined value.

Pauses the workflow for manual intervention.

When called inside a step (workflow state is 'performing'): interrupts execution via throw/catch. When called outside a step (workflow state is 'ready'): directly pauses the workflow.

Can be resumed later with Workflow#resume!.

Raises:



137
138
139
140
141
142
143
144
 
# File 'lib/geneva_drive/flow_control.rb', line 137

def pause!
  if state == "performing"
    logger.info("Flow control: pause! called from step")
    throw :flow_control, GenevaDrive::FlowControlSignal.new(:pause)
  else
    external_pause!
  end
end

#reattempt!(wait: nil, rewind: false) ⇒ void

This method returns an undefined value.

Reschedules the current step for another attempt. Useful for handling temporary failures or rate limiting.

For resumable steps, by default continues from the current cursor position. Use rewind: true to restart the step from the beginning.

Examples:

Retry after rate limit

reattempt!(wait: 5.minutes)

Retry resumable step from the beginning

reattempt!(rewind: true)

Parameters:

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

    optional delay before retry

  • rewind (Boolean) (defaults to: false) —

    if true, resets cursor to beginning (resumable steps only)

Raises:

  • (UncaughtThrowError) —

    if called outside of step execution context



162
163
164
165
166
167
 
# File 'lib/geneva_drive/flow_control.rb', line 162

def reattempt!(wait: nil, rewind: false)
  wait_msg = wait ? " with wait #{wait.inspect}" : ""
  rewind_msg = rewind ? " (rewinding cursor)" : ""
  logger.info("Flow control: reattempt! called from step#{wait_msg}#{rewind_msg}")
  throw :flow_control, GenevaDrive::FlowControlSignal.new(:reattempt, wait: wait, rewind: rewind)
end

#skip! ⇒ void

This method returns an undefined value.

Skips the current step and proceeds to the next one.

When called inside a step (workflow state is 'performing'): interrupts execution via throw/catch. When called outside a step (workflow state is 'ready'): directly skips current step.

Raises:



176
177
178
179
180
181
182
183
 
# File 'lib/geneva_drive/flow_control.rb', line 176

def skip!
  if state == "performing"
    logger.info("Flow control: skip! called from step")
    throw :flow_control, GenevaDrive::FlowControlSignal.new(:skip)
  else
    external_skip!
  end
end

#suspend!(wait: nil) ⇒ void

This method returns an undefined value.

Suspends the current resumable step and re-enqueues for later continuation. The step will resume from its current cursor position.

This is primarily for use within resumable steps when you need to explicitly yield control (e.g., to respect rate limits or system load).

Examples:

Suspend for rate limiting

suspend!(wait: 5.minutes)

Suspend to yield control

suspend!

Parameters:

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

    optional delay before re-enqueue

Raises:

  • (UncaughtThrowError) —

    if called outside of step execution context



210
211
212
213
214
215
216
217
218
219
220
 
# File 'lib/geneva_drive/flow_control.rb', line 210

def suspend!(wait: nil)
  executing_step = current_step_name && steps.named(current_step_name)
  unless executing_step&.resumable?
    raise GenevaDrive::InvalidStateError,
      "suspend! can only be called from inside a resumable_step (it continues from the persisted cursor)"
  end

  wait_msg = wait ? " with wait #{wait.inspect}" : ""
  logger.info("Flow control: suspend! called from step#{wait_msg}")
  throw :flow_control, GenevaDrive::FlowControlSignal.new(:suspend, wait: wait)
end