Class: Fractor::Workflow

Inherits:
Object
  • Object
show all
Defined in:
lib/fractor/workflow.rb,
lib/fractor/workflow/job.rb,
lib/fractor/workflow/logger.rb,
lib/fractor/workflow/builder.rb,
lib/fractor/workflow/helpers.rb,
lib/fractor/workflow/visualizer.rb,
lib/fractor/workflow/retry_config.rb,
lib/fractor/workflow/chain_builder.rb,
lib/fractor/workflow/retry_strategy.rb,
lib/fractor/workflow/circuit_breaker.rb,
lib/fractor/workflow/execution_hooks.rb,
lib/fractor/workflow/execution_trace.rb,
lib/fractor/workflow/workflow_result.rb,
lib/fractor/workflow/workflow_context.rb,
lib/fractor/workflow/dead_letter_queue.rb,
lib/fractor/workflow/structured_logger.rb,
lib/fractor/workflow/workflow_executor.rb,
lib/fractor/workflow/execution_strategy.rb,
lib/fractor/workflow/retry_orchestrator.rb,
lib/fractor/workflow/workflow_validator.rb,
lib/fractor/workflow/pre_execution_context.rb,
lib/fractor/workflow/execution/job_executor.rb,
lib/fractor/workflow/circuit_breaker_registry.rb,
lib/fractor/workflow/execution/result_builder.rb,
lib/fractor/workflow/job_dependency_validator.rb,
lib/fractor/workflow/circuit_breaker_orchestrator.rb,
lib/fractor/workflow/type_compatibility_validator.rb,
lib/fractor/workflow/execution/dependency_resolver.rb,
lib/fractor/workflow/execution/fallback_job_handler.rb,
lib/fractor/workflow/execution/workflow_execution_logger.rb

Overview

Base class for defining workflows using a declarative DSL. Workflows coordinate multiple jobs with dependencies, type-safe data flow, and support both pipeline and continuous execution modes.

Defined Under Namespace

Modules: Helpers Classes: Builder, ChainBuilder, CircuitBreaker, CircuitBreakerOrchestrator, CircuitBreakerRegistry, CircuitOpenError, ConstantDelay, DeadLetterQueue, DependencyResolver, ExecutionHooks, ExecutionStrategy, ExecutionTrace, ExponentialBackoff, FallbackJobHandler, FilePersister, Job, JobDependencyValidator, JobExecutor, LinearBackoff, NoRetry, ParallelExecutionStrategy, PipelineExecutionStrategy, PreExecutionContext, ResultBuilder, RetryConfig, RetryOrchestrator, RetryState, RetryStrategy, SequentialExecutionStrategy, StructuredLogger, TypeCompatibilityValidator, Visualizer, WorkflowContext, WorkflowExecutionLogger, WorkflowExecutor, WorkflowLogger, WorkflowResult, WorkflowValidator

Class Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(input = nil) ⇒ Workflow

Create a new workflow instance.



248
249
250
251
252
253
254
255
 
# File 'lib/fractor/workflow.rb', line 248

def initialize(input = nil)
  unless self.class.workflow_name
    raise "Workflow not defined. Use 'workflow \"name\" do ... end' in class definition"
  end

  @workflow_input = input
  @dead_letter_queue = initialize_dead_letter_queue
end

Class Attribute Details

.dlq_config ⇒ Object (readonly)

Returns the value of attribute dlq_config.



28
29
30
 
# File 'lib/fractor/workflow.rb', line 28

def dlq_config
  @dlq_config
end

.end_job_names ⇒ Object (readonly)

Returns the value of attribute end_job_names.



28
29
30
 
# File 'lib/fractor/workflow.rb', line 28

def end_job_names
  @end_job_names
end

.input_model_class ⇒ Object (readonly)

Returns the value of attribute input_model_class.



28
29
30
 
# File 'lib/fractor/workflow.rb', line 28

def input_model_class
  @input_model_class
end

.jobs ⇒ Object (readonly)

Returns the value of attribute jobs.



28
29
30
 
# File 'lib/fractor/workflow.rb', line 28

def jobs
  @jobs
end

.output_model_class ⇒ Object (readonly)

Returns the value of attribute output_model_class.



28
29
30
 
# File 'lib/fractor/workflow.rb', line 28

def output_model_class
  @output_model_class
end

.start_job_name ⇒ Object (readonly)

Returns the value of attribute start_job_name.



28
29
30
 
# File 'lib/fractor/workflow.rb', line 28

def start_job_name
  @start_job_name
end

.workflow_mode ⇒ Object (readonly)

Returns the value of attribute workflow_mode.



28
29
30
 
# File 'lib/fractor/workflow.rb', line 28

def workflow_mode
  @workflow_mode
end

.workflow_name ⇒ Object (readonly)

Returns the value of attribute workflow_name.



28
29
30
 
# File 'lib/fractor/workflow.rb', line 28

def workflow_name
  @workflow_name
end

Class Method Details

.chain(name) ⇒ ChainBuilder

Create a linear chain workflow for sequential processing. This provides a fluent API for simple pipelines.

Examples:

workflow = Fractor::Workflow.chain("pipeline")
  .step("uppercase", UppercaseWorker)
  .step("reverse", ReverseWorker)
  .step("finalize", FinalizeWorker)
  .build

Parameters:

  • name (String) —

    The workflow name

Returns:



66
67
68
 
# File 'lib/fractor/workflow.rb', line 66

def chain(name)
  ChainBuilder.new(name)
end

.configure_dead_letter_queue(max_size: 1000, persister: nil, on_add: nil) ⇒ Object

Configure the Dead Letter Queue for failed work.

Parameters:

  • max_size (Integer) (defaults to: 1000) —

    Maximum number of entries to retain

  • persister (Object) (defaults to: nil) —

    Optional persistence strategy

  • on_add (Proc) (defaults to: nil) —

    Optional callback when entry is added



126
127
128
129
130
131
132
133
 
# File 'lib/fractor/workflow.rb', line 126

def configure_dead_letter_queue(max_size: 1000, persister: nil,
on_add: nil)
  @dlq_config = {
    max_size: max_size,
    persister: persister,
    on_add: on_add,
  }
end

.define(name, mode: :pipeline) { ... } ⇒ Class

Create a workflow class without inheritance. This is a convenience method that creates an anonymous workflow class.

Examples:

workflow = Fractor::Workflow.define("my-workflow") do
  job "step1", Worker1
  job "step2", Worker2, needs: "step1"
  job "step3", Worker3, needs: "step2"
end
instance = workflow.new
result = instance.execute(input: data)

Parameters:

  • name (String) —

    The workflow name

  • mode (Symbol) (defaults to: :pipeline) —

    :pipeline (default) or :continuous

Yields:

  • Block containing job definitions using workflow DSL

Returns:

  • (Class) —

    A new Workflow subclass



48
49
50
51
52
 
# File 'lib/fractor/workflow.rb', line 48

def define(name, mode: :pipeline, &block)
  Class.new(Workflow) do
    workflow(name, mode: mode, &block)
  end
end

.end_with(job_name, on: :success) ⇒ Object

Define an ending job for pipeline mode.

Parameters:

  • job_name (String, Symbol) —

    The name of the end job

  • on (Symbol) (defaults to: :success) —

    Condition: :success (default), :failure, :cancellation



117
118
119
 
# File 'lib/fractor/workflow.rb', line 117

def end_with(job_name, on: :success)
  @end_job_names << { name: job_name.to_s, condition: on }
end

.input_type(klass) ⇒ Object

Declare the workflow's input type.

Parameters:

  • klass (Class) —

    A Lutaml::Model::Serializable subclass



93
94
95
96
 
# File 'lib/fractor/workflow.rb', line 93

def input_type(klass)
  validate_model_class!(klass, "input_type")
  @input_model_class = klass
end

.job(name, worker_class = nil, needs: nil, inputs: nil, outputs: nil, workers: nil, condition: nil) { ... } ⇒ Object

Define a job in the workflow.

Examples:

DSL syntax (original)

job "process" do
  runs_with ProcessWorker
  needs "validate"
end

Shorthand syntax (simplified)

job "process", ProcessWorker, needs: "validate"

Shorthand with multiple options

job "process", ProcessWorker, needs: "validate", outputs: :workflow

Parameters:

  • name (String, Symbol) —

    The job name

  • worker_class (Class) (defaults to: nil) —

    Optional worker class (shorthand syntax)

  • needs (String, Symbol, Array) (defaults to: nil) —

    Optional dependencies (shorthand)

  • inputs (Symbol, String, Hash) (defaults to: nil) —

    Optional input configuration (shorthand)

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

    Optional :workflow to mark outputs (shorthand)

  • workers (Integer) (defaults to: nil) —

    Optional parallel worker count (shorthand)

  • condition (Proc) (defaults to: nil) —

    Optional conditional execution (shorthand)

Yields:

  • Block containing job configuration (DSL syntax)



157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/fractor/workflow.rb', line 157

def job(name, worker_class = nil, needs: nil, inputs: nil, outputs: nil,
        workers: nil, condition: nil, &block)
  job_name = name.to_s
  if @jobs.key?(job_name)
    raise ArgumentError,
          "Job '#{job_name}' already defined"
  end

  job_obj = Job.new(job_name, self)

  # Apply shorthand parameters if provided
  if worker_class
    job_obj.runs_with(worker_class)
  end

  if needs
    needs_array = needs.is_a?(Array) ? needs : [needs]
    job_obj.needs(*needs_array)
  end

  if inputs
    case inputs
    when :workflow, "workflow"
      job_obj.inputs_from_workflow
    when String, Symbol
      job_obj.inputs_from_job(inputs.to_s)
    when Hash
      job_obj.inputs_from_multiple(inputs)
    end
  end

  if outputs == :workflow
    job_obj.outputs_to_workflow
  end

  if workers
    job_obj.parallel_workers(workers)
  end

  if condition
    job_obj.if_condition(condition)
  end

  # Apply DSL block if provided
  job_obj.instance_eval(&block) if block

  @jobs[job_name] = job_obj
end

.output_type(klass) ⇒ Object

Declare the workflow's output type.

Parameters:

  • klass (Class) —

    A Lutaml::Model::Serializable subclass



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

def output_type(klass)
  validate_model_class!(klass, "output_type")
  @output_model_class = klass
end

.print_diagram ⇒ Object

Print ASCII diagram to stdout



228
229
230
 
# File 'lib/fractor/workflow.rb', line 228

def print_diagram
  Visualizer.new(self).print
end

.start_with(job_name) ⇒ Object

Define the starting job for pipeline mode.

Parameters:

  • job_name (String, Symbol) —

    The name of the start job



109
110
111
 
# File 'lib/fractor/workflow.rb', line 109

def start_with(job_name)
  @start_job_name = job_name.to_s
end

.to_ascii ⇒ String

Generate an ASCII art diagram of the workflow

Returns:

  • (String) —

    ASCII art representation



223
224
225
 
# File 'lib/fractor/workflow.rb', line 223

def to_ascii
  Visualizer.new(self).to_ascii
end

.to_dot ⇒ String

Generate a DOT/Graphviz diagram of the workflow

Returns:

  • (String) —

    DOT diagram syntax



216
217
218
 
# File 'lib/fractor/workflow.rb', line 216

def to_dot
  Visualizer.new(self).to_dot
end

.to_mermaid ⇒ String

Generate a Mermaid flowchart diagram of the workflow

Returns:

  • (String) —

    Mermaid diagram syntax



209
210
211
 
# File 'lib/fractor/workflow.rb', line 209

def to_mermaid
  Visualizer.new(self).to_mermaid
end

.workflow(name, mode: :pipeline) { ... } ⇒ Object

Define a workflow with the given name and optional mode.

Parameters:

  • name (String) —

    The workflow name

  • mode (Symbol) (defaults to: :pipeline) —

    :pipeline (default) or :continuous

Yields:

  • Block containing job definitions



75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/fractor/workflow.rb', line 75

def workflow(name, mode: :pipeline, &block)
  @workflow_name = name
  @workflow_mode = mode
  @jobs = {}
  @start_job_name = nil
  @end_job_names = []
  @input_model_class = nil
  @output_model_class = nil
  @dlq_config = nil

  instance_eval(&block) if block

  validate_workflow!
end

Instance Method Details

#dead_letter_queue ⇒ DeadLetterQueue?

Access the Dead Letter Queue for this workflow.

Returns:



260
261
262
 
# File 'lib/fractor/workflow.rb', line 260

def dead_letter_queue
  @dead_letter_queue
end

#execute(input: nil, correlation_id: nil, logger: nil, trace: false) {|WorkflowExecutor| ... } ⇒ WorkflowResult

Execute the workflow with the given input.

Parameters:

  • input (Lutaml::Model::Serializable, nil) (defaults to: nil) —

    The workflow input (optional if provided to initialize)

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

    Optional correlation ID for tracking

  • logger (Logger) (defaults to: nil) —

    Optional logger instance

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

    Whether to generate execution trace

Yields:

Returns:



272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
 
# File 'lib/fractor/workflow.rb', line 272

def execute(input: nil, correlation_id: nil, logger: nil, trace: false,
&block)
  # Use provided input or fall back to initialization input
  workflow_input = input || @workflow_input
  validate_input!(workflow_input)

  executor = WorkflowExecutor.new(
    self,
    workflow_input,
    correlation_id: correlation_id,
    logger: logger,
    trace: trace,
    dead_letter_queue: @dead_letter_queue,
  )

  # Allow block to register hooks
  block&.call(executor)

  executor.execute
end

#run_continuous(work_queue:) ⇒ Object

Run the workflow in continuous mode with a work queue.

Parameters:

  • work_queue (WorkQueue) —

    The queue to receive workflow inputs

Raises:

  • (NotImplementedError) 


296
297
298
299
300
301
302
303
 
# File 'lib/fractor/workflow.rb', line 296

def run_continuous(work_queue:)
  unless self.class.workflow_mode == :continuous
    raise "Workflow '#{self.class.workflow_name}' is not configured for continuous mode"
  end

  # Continuous mode implementation will be added
  raise NotImplementedError, "Continuous mode coming soon"
end