Class: Fractor::Workflow::Job

Inherits:
Object
  • Object
show all
Defined in:
lib/fractor/workflow/job.rb

Overview

Represents a single job in a workflow. Jobs encapsulate worker configuration, dependencies, and input/output mappings.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name, workflow_class) ⇒ Job

Returns a new instance of Job.



15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/fractor/workflow/job.rb', line 15

def initialize(name, workflow_class)
  @name = name
  @workflow_class = workflow_class
  @worker_class = nil
  @dependencies = []
  @num_workers = nil
  @input_mappings = {}
  @condition_proc = nil
  @terminates = false
  @outputs_to_workflow = false
  @state = :pending
  @retry_config = nil
  @error_handlers = []
  @fallback_job = nil
  @circuit_breaker_config = nil
end

Instance Attribute Details

#circuit_breaker_config ⇒ Object (readonly)

Returns the value of attribute circuit_breaker_config.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def circuit_breaker_config
  @circuit_breaker_config
end

#condition_proc ⇒ Object (readonly)

Returns the value of attribute condition_proc.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def condition_proc
  @condition_proc
end

#dependencies ⇒ Object (readonly)

Returns the value of attribute dependencies.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def dependencies
  @dependencies
end

#error_handlers ⇒ Object (readonly)

Returns the value of attribute error_handlers.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def error_handlers
  @error_handlers
end

#fallback_job ⇒ Object (readonly)

Returns the value of attribute fallback_job.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def fallback_job
  @fallback_job
end

#input_mappings ⇒ Object (readonly)

Returns the value of attribute input_mappings.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def input_mappings
  @input_mappings
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def name
  @name
end

#num_workers ⇒ Object (readonly)

Returns the value of attribute num_workers.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def num_workers
  @num_workers
end

#retry_config ⇒ Object (readonly)

Returns the value of attribute retry_config.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def retry_config
  @retry_config
end

#terminates ⇒ Object (readonly)

Returns the value of attribute terminates.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def terminates
  @terminates
end

#worker_class ⇒ Object (readonly)

Returns the value of attribute worker_class.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def worker_class
  @worker_class
end

#workflow_class ⇒ Object (readonly)

Returns the value of attribute workflow_class.



10
11
12
 
# File 'lib/fractor/workflow/job.rb', line 10

def workflow_class
  @workflow_class
end

Instance Method Details

#auto_wire_inputs! ⇒ Object

Auto-wire inputs from dependencies if not explicitly configured. Called during workflow finalization.



70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/fractor/workflow/job.rb', line 70

def auto_wire_inputs!
  return unless @input_mappings.empty?

  if @dependencies.empty?
    # No dependencies = must be a start job
    @input_mappings[:workflow] = true
  elsif @dependencies.size == 1
    # Single dependency = auto-wire from that job
    @input_mappings[@dependencies.first] = :all
  end
  # Multiple dependencies require explicit configuration
end

#circuit_breaker(threshold: 5, timeout: 60, half_open_calls: 3, shared_key: nil) ⇒ Object

Configure circuit breaker for this job.

Parameters:

  • threshold (Integer) (defaults to: 5) —

    Number of failures before opening circuit

  • timeout (Integer) (defaults to: 60) —

    Seconds to wait before trying half-open

  • half_open_calls (Integer) (defaults to: 3) —

    Number of test calls in half-open

  • shared_key (String) (defaults to: nil) —

    Optional key for shared circuit breaker



178
179
180
181
182
183
184
185
186
 
# File 'lib/fractor/workflow/job.rb', line 178

def circuit_breaker(threshold: 5, timeout: 60, half_open_calls: 3,
                    shared_key: nil)
  @circuit_breaker_config = {
    threshold: threshold,
    timeout: timeout,
    half_open_calls: half_open_calls,
    shared_key: shared_key,
  }
end

#circuit_breaker_enabled? ⇒ Boolean

Check if this job has circuit breaker configured.

Returns:

  • (Boolean) —

    Whether circuit breaker is configured



191
192
193
 
# File 'lib/fractor/workflow/job.rb', line 191

def circuit_breaker_enabled?
  !@circuit_breaker_config.nil?
end

#circuit_breaker_key ⇒ String

Get the circuit breaker key for this job.

Returns:

  • (String) —

    The circuit breaker key



198
199
200
201
202
 
# File 'lib/fractor/workflow/job.rb', line 198

def circuit_breaker_key
  return nil unless circuit_breaker_enabled?

  @circuit_breaker_config[:shared_key] || "job_#{@name}"
end

#fallback_to(job_name) ⇒ Object

Set a fallback job for this job.

Parameters:

  • job_name (String, Symbol) —

    Name of the fallback job



168
169
170
 
# File 'lib/fractor/workflow/job.rb', line 168

def fallback_to(job_name)
  @fallback_job = job_name.to_s
end

#handle_error(error, context) ⇒ Object

Execute error handlers for this job.

Parameters:

  • error (Exception) —

    The error that occurred

  • context (WorkflowContext) —

    The workflow context



222
223
224
225
226
227
228
229
230
 
# File 'lib/fractor/workflow/job.rb', line 222

def handle_error(error, context)
  @error_handlers.each do |handler|
    handler.call(error, context)
  rescue StandardError => e
    context.logger&.error(
      "Error handler failed for job #{@name}: #{e.message}",
    )
  end
end

#if_condition(proc) ⇒ Object

Set a condition for this job to run.

Parameters:

  • proc (Proc) —

    A proc that receives the workflow context



110
111
112
113
114
115
116
 
# File 'lib/fractor/workflow/job.rb', line 110

def if_condition(proc)
  unless proc.respond_to?(:call)
    raise ArgumentError, "if_condition must be callable"
  end

  @condition_proc = proc
end

#input_type ⇒ Class

Get the input type for this job from its worker.

Returns:

  • (Class) —

    The input type class



252
253
254
255
256
 
# File 'lib/fractor/workflow/job.rb', line 252

def input_type
  return nil unless @worker_class

  @worker_class.input_type_class
end

#inputs_from_job(source_job, select: nil) ⇒ Object

Map inputs from a single upstream job.

Parameters:

  • source_job (String, Symbol) —

    The source job name

  • select (Hash) (defaults to: nil) —

    Optional attribute mappings



87
88
89
90
 
# File 'lib/fractor/workflow/job.rb', line 87

def inputs_from_job(source_job, select: nil)
  source = source_job.to_s
  @input_mappings[source] = select || :all
end

#inputs_from_multiple(mappings) ⇒ Object

Map inputs from multiple upstream jobs.

Example:

inputs_from_multiple(
"job_a" => { validated_data: :validated_data },
"job_b" => { analysis: :results }
)

Parameters:

  • mappings (Hash) —

    Hash of source_job => attribute_mappings



100
101
102
103
104
105
 
# File 'lib/fractor/workflow/job.rb', line 100

def inputs_from_multiple(mappings)
  mappings.each do |source_job, attr_mappings|
    source = source_job.to_s
    @input_mappings[source] = attr_mappings
  end
end

#inputs_from_workflow ⇒ Object Also known as: inputs_from

Map inputs from the workflow input. Used when this is the first job in the workflow.



63
64
65
 
# File 'lib/fractor/workflow/job.rb', line 63

def inputs_from_workflow
  @input_mappings[:workflow] = true
end

#needs(*job_names) ⇒ Object

Specify job dependencies.

Parameters:

  • job_names (Array<String, Symbol>) —

    Names of jobs this job depends on



46
47
48
 
# File 'lib/fractor/workflow/job.rb', line 46

def needs(*job_names)
  @dependencies = job_names.flatten.map(&:to_s)
end

#on_error(&handler) ⇒ Object

Add an error handler for this job.

Parameters:

  • handler (Proc) —

    A proc that receives (error, context)



157
158
159
160
161
162
163
 
# File 'lib/fractor/workflow/job.rb', line 157

def on_error(&handler)
  unless handler.respond_to?(:call)
    raise ArgumentError, "on_error must be given a block"
  end

  @error_handlers << handler
end

#output_type ⇒ Class

Get the output type for this job from its worker.

Returns:

  • (Class) —

    The output type class



261
262
263
264
265
 
# File 'lib/fractor/workflow/job.rb', line 261

def output_type
  return nil unless @worker_class

  @worker_class.output_type_class
end

#outputs_to_workflow ⇒ Object Also known as: outputs_to

Mark this job's outputs as mapping to workflow outputs.



126
127
128
 
# File 'lib/fractor/workflow/job.rb', line 126

def outputs_to_workflow
  @outputs_to_workflow = true
end

#outputs_to_workflow? ⇒ Boolean

Check if this job's outputs map to workflow outputs.

Returns:

  • (Boolean) —

    Whether this job outputs to workflow



235
236
237
 
# File 'lib/fractor/workflow/job.rb', line 235

def outputs_to_workflow?
  @outputs_to_workflow
end

#parallel_workers(n) ⇒ Object

Set the number of parallel workers for this job.

Parameters:

  • n (Integer) —

    Number of workers



53
54
55
56
57
58
59
 
# File 'lib/fractor/workflow/job.rb', line 53

def parallel_workers(n)
  unless n.is_a?(Integer) && n.positive?
    raise ArgumentError, "parallel_workers must be a positive integer"
  end

  @num_workers = n
end

#ready?(completed_jobs) ⇒ Boolean

Check if this job is ready to execute. A job is ready when all its dependencies have completed.

Parameters:

  • completed_jobs (Set) —

    Set of completed job names

Returns:

  • (Boolean) —

    Whether the job is ready



272
273
274
 
# File 'lib/fractor/workflow/job.rb', line 272

def ready?(completed_jobs)
  @dependencies.all? { |dep| completed_jobs.include?(dep) }
end

#retry_enabled? ⇒ Boolean

Check if this job has retry configured.

Returns:

  • (Boolean) —

    Whether retry is configured



207
208
209
 
# File 'lib/fractor/workflow/job.rb', line 207

def retry_enabled?
  !@retry_config.nil? && @retry_config.max_attempts > 1
end

#retry_on_error(max_attempts: 3, backoff: :exponential, initial_delay: 1, max_delay: nil, timeout: nil, retryable_errors: [StandardError], **options) ⇒ Object

Configure retry behavior for this job.

Parameters:

  • max_attempts (Integer) (defaults to: 3) —

    Maximum number of retry attempts

  • backoff (Symbol) (defaults to: :exponential) —

    Backoff strategy (:exponential, :linear, :constant, :none)

  • initial_delay (Numeric) (defaults to: 1) —

    Initial delay in seconds

  • max_delay (Numeric) (defaults to: nil) —

    Maximum delay in seconds

  • timeout (Numeric) (defaults to: nil) —

    Job execution timeout in seconds

  • retryable_errors (Array<Class>) (defaults to: [StandardError]) —

    List of retryable error classes

  • options (Hash) —

    Additional retry options



140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/fractor/workflow/job.rb', line 140

def retry_on_error(max_attempts: 3, backoff: :exponential,
                   initial_delay: 1, max_delay: nil, timeout: nil,
                   retryable_errors: [StandardError], **options)
  @retry_config = Workflow::RetryConfig.from_options(
    max_attempts: max_attempts,
    backoff: backoff,
    initial_delay: initial_delay,
    max_delay: max_delay,
    timeout: timeout,
    retryable_errors: retryable_errors,
    **options,
  )
end

#runs_with(klass) ⇒ Object

Specify which worker class processes this job.

Parameters:

  • klass (Class) —

    A Fractor::Worker subclass



35
36
37
38
39
40
41
 
# File 'lib/fractor/workflow/job.rb', line 35

def runs_with(klass)
  unless klass < Fractor::Worker
    raise ArgumentError, "#{klass} must inherit from Fractor::Worker"
  end

  @worker_class = klass
end

#should_execute?(context) ⇒ Boolean

Check if this job should execute based on its condition.

Parameters:

Returns:

  • (Boolean) —

    Whether the job should execute



243
244
245
246
247
 
# File 'lib/fractor/workflow/job.rb', line 243

def should_execute?(context)
  return true unless @condition_proc

  @condition_proc.call(context)
end

#state(new_state = nil) ⇒ Symbol

Get or set the job state.

Parameters:

  • new_state (Symbol) (defaults to: nil) —

    :pending, :ready, :running, :completed, :failed, :skipped

Returns:

  • (Symbol) —

    The current state



280
281
282
283
 
# File 'lib/fractor/workflow/job.rb', line 280

def state(new_state = nil)
  @state = new_state if new_state
  @state
end

#terminates_workflow(value = true) ⇒ Object

Mark this job as a workflow terminator.

Parameters:

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

    Whether this job terminates the workflow



121
122
123
 
# File 'lib/fractor/workflow/job.rb', line 121

def terminates_workflow(value = true)
  @terminates = value
end

#timeout ⇒ Numeric?

Get the timeout for this job.

Returns:

  • (Numeric, nil) —

    Timeout in seconds or nil



214
215
216
 
# File 'lib/fractor/workflow/job.rb', line 214

def timeout
  @retry_config&.timeout
end

#to_s ⇒ Object 



285
286
287
 
# File 'lib/fractor/workflow/job.rb', line 285

def to_s
  "Job[#{@name}]"
end