Class: GenevaDrive::IterableStep

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

Overview

Object passed to resumable step blocks for cursor-based iteration.

Provides an API compatible with Rails 8.1's ActiveJob::Continuation::Step, plus GenevaDrive-specific extensions for iterating over records and batches.

Examples:

Manual cursor control

resumable_step :process_records do |iter|
  hero.records.where("id > ?", iter.cursor || 0).find_each do |record|
    process(record)
    iter.set!(record.id)
  end
end

Using iterate_over_records helper

resumable_step :process_records do |iter|
  iter.iterate_over_records(hero.records) do |record|
    process(record)
  end
end

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name, cursor, execution:, resumed:, interrupter:) ⇒ IterableStep

Creates a new IterableStep for a resumable step execution.

Parameters:

  • name (String, Symbol) —

    the step name

  • cursor (Object, nil) —

    the current cursor value

  • execution (GenevaDrive::StepExecution) —

    the step execution record

  • resumed (Boolean) —

    whether this is a resumed execution

  • interrupter (#should_interrupt?) —

    object to check for interruption



33
34
35
36
37
38
39
40
41
 
# File 'lib/geneva_drive/iterable_step.rb', line 33

def initialize(name, cursor, execution:, resumed:, interrupter:)
  @name = name.to_sym
  @cursor = cursor
  @execution = execution
  @resumed = resumed
  @interrupter = interrupter
  @advanced = false
  @iterations = 0
end

Instance Attribute Details

#cursor ⇒ Object (readonly)

Returns the value of attribute cursor.



24
25
26
 
# File 'lib/geneva_drive/iterable_step.rb', line 24

def cursor
  @cursor
end

#iterations ⇒ Integer (readonly)

Returns the number of iterations completed during this execution (checkpoints via set!, advance! or checkpoint!).

Returns:

  • (Integer) 


47
48
49
 
# File 'lib/geneva_drive/iterable_step.rb', line 47

def iterations
  @iterations
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



24
25
26
 
# File 'lib/geneva_drive/iterable_step.rb', line 24

def name
  @name
end

Instance Method Details

#advance! ⇒ void

This method returns an undefined value.

Increments an integer cursor by 1 and checkpoints. Only works with integer cursors.

Raises:

  • (ArgumentError) —

    if cursor is not an Integer



81
82
83
84
 
# File 'lib/geneva_drive/iterable_step.rb', line 81

def advance!
  raise ArgumentError, "advance! requires integer cursor, got #{@cursor.inspect}" unless @cursor.is_a?(Integer)
  set!(@cursor + 1)
end

#advanced? ⇒ Boolean

Returns true if the cursor was changed during this execution.

Returns:

  • (Boolean) 


61
62
63
 
# File 'lib/geneva_drive/iterable_step.rb', line 61

def advanced?
  @advanced
end

#checkpoint! ⇒ void

This method returns an undefined value.

Persists the current cursor, counts the iteration, and checks for interruption (max_iterations, max_runtime, shutdown, external pause/cancel). Called automatically by set! and advance!.



91
92
93
94
95
 
# File 'lib/geneva_drive/iterable_step.rb', line 91

def checkpoint!
  @iterations += 1
  persist_cursor!
  check_interruption!
end

#description ⇒ String

Returns a human-readable description of current position.

Returns:

  • (String) 


107
108
109
 
# File 'lib/geneva_drive/iterable_step.rb', line 107

def description
  "at '#{name}', cursor '#{cursor.inspect}'"
end

#iterate_over(enumerable) {|Object| ... } ⇒ void

This method returns an undefined value.

Iterates over an enumerable using index as cursor. Resumes from the stored cursor position on subsequent runs.

Parameters:

  • enumerable (Enumerable) —

    the collection to iterate over

Yields:

  • (Object) —

    each element in the collection

Raises:

  • (ArgumentError) 


132
133
134
135
136
137
138
139
140
141
 
# File 'lib/geneva_drive/iterable_step.rb', line 132

def iterate_over(enumerable)
  raise ArgumentError, "Must respond to #each" unless enumerable.respond_to?(:each)

  start_index = @cursor || 0
  enumerable.each_with_index do |element, index|
    next if index < start_index
    yield element
    set!(index + 1)
  end
end

#iterate_over_records(relation, cursor: :id) {|ActiveRecord::Base| ... } ⇒ void

This method returns an undefined value.

Iterates over an ActiveRecord relation using a column as cursor. Uses find_each for memory-efficient batching.

Parameters:

  • relation (ActiveRecord::Relation) —

    the relation to iterate over

  • cursor (Symbol) (defaults to: :id) —

    the column to use as cursor (default: :id)

Yields:

  • (ActiveRecord::Base) —

    each record in the relation



150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/geneva_drive/iterable_step.rb', line 150

def iterate_over_records(relation, cursor: :id)
  filtered = if @cursor
    relation.where(relation.arel_table[cursor].gt(@cursor)).order(cursor => :asc)
  else
    relation.order(cursor => :asc)
  end

  filtered.find_each do |record|
    yield record
    set!(record.public_send(cursor))
  end
end

#iterate_over_subrelations(relation, batch_size:, cursor: :id) {|ActiveRecord::Relation| ... } ⇒ void

This method returns an undefined value.

Iterates over an ActiveRecord relation in subrelations (batches). Each subrelation is yielded as an ActiveRecord relation with offset applied, useful for bulk operations like updates or exports.

Parameters:

  • relation (ActiveRecord::Relation) —

    the relation to iterate over

  • batch_size (Integer) —

    the number of records per subrelation

  • cursor (Symbol) (defaults to: :id) —

    the column to use as cursor (default: :id)

Yields:

  • (ActiveRecord::Relation) —

    each subrelation with offset applied



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

def iterate_over_subrelations(relation, batch_size:, cursor: :id)
  filtered = if @cursor
    relation.where(relation.arel_table[cursor].gt(@cursor)).order(cursor => :asc)
  else
    relation.order(cursor => :asc)
  end

  filtered.in_batches(of: batch_size) do |relation_with_offset|
    yield relation_with_offset
    last_cursor = relation_with_offset.maximum(cursor)
    set!(last_cursor) if last_cursor
  end
end

#resumed? ⇒ Boolean

Returns true if this execution is resuming from a prior suspension.

Returns:

  • (Boolean) 


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

def resumed?
  @resumed
end

#set!(value) ⇒ void

This method returns an undefined value.

Sets the cursor to a new value and checkpoints. The cursor can be any value that ActiveJob can serialize.

Parameters:

  • value (Object) —

    the new cursor value



70
71
72
73
74
 
# File 'lib/geneva_drive/iterable_step.rb', line 70

def set!(value)
  @cursor = value
  @advanced = true
  checkpoint!
end

#skip_to!(value, wait: nil) ⇒ void

This method returns an undefined value.

Skips to a new cursor position and suspends execution. The job will be re-enqueued to continue from the new cursor.

Parameters:

  • value (Object) —

    the new cursor value

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

    optional delay before re-enqueue



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

def skip_to!(value, wait: nil)
  @cursor = value
  @advanced = true
  persist_cursor!
  throw :interrupt, wait
end

#to_a ⇒ Array

Returns array representation [name, cursor] for compatibility.

Returns:

  • (Array) 


100
101
102
 
# File 'lib/geneva_drive/iterable_step.rb', line 100

def to_a
  [name.to_s, cursor]
end