Class: Fractor::Workflow::CircuitBreakerOrchestrator

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

Overview

Orchestrates circuit breaker logic for workflow job execution. Wraps a CircuitBreaker and provides workflow-specific integration.

Examples:

Basic usage

orchestrator = CircuitBreakerOrchestrator.new(threshold: 5, timeout: 60)
result = orchestrator.execute_with_breaker(job) { |job| execute_job(job) }

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(threshold: 5, timeout: 60, half_open_calls: 3, job_name: nil, debug: false) ⇒ CircuitBreakerOrchestrator

Initialize a new circuit breaker orchestrator.

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

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

    Optional job name for logging

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

    Whether to enable debug logging



23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 23

def initialize(threshold: 5, timeout: 60, half_open_calls: 3,
job_name: nil, debug: false)
  @breaker = CircuitBreaker.new(
    threshold: threshold,
    timeout: timeout,
    half_open_calls: half_open_calls,
  )
  @job_name = job_name
  @debug = debug
  @execution_count = 0
  @success_count = 0
  @blocked_count = 0
end

Instance Attribute Details

#breaker ⇒ Object (readonly)

Returns the value of attribute breaker.



14
15
16
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 14

def breaker
  @breaker
end

#debug ⇒ Object (readonly)

Returns the value of attribute debug.



14
15
16
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 14

def debug
  @debug
end

#job_name ⇒ Object (readonly)

Returns the value of attribute job_name.



14
15
16
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 14

def job_name
  @job_name
end

Instance Method Details

#close_circuit! ⇒ Object

Manually close the circuit (for testing or recovery).



166
167
168
169
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 166

def close_circuit!
  @breaker.reset
  log_debug "Circuit manually closed for job '#{@job_name}'"
end

#closed? ⇒ Boolean

Check if the circuit is currently closed.

Returns:

  • (Boolean) —

    true if circuit is closed



68
69
70
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 68

def closed?
  @breaker.closed?
end

#execute_bypassing_breaker(job) {|Job| ... } ⇒ Object

Try to execute the job regardless of circuit state. This bypasses the circuit breaker but still tracks results.

Parameters:

  • job (Job) —

    The job to execute

Yields:

  • (Job) —

    Block that executes the job

Returns:

  • (Object) —

    The execution result



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

def execute_bypassing_breaker(job)
  @execution_count += 1

  log_debug "Executing job '#{job.name}' bypassing circuit breaker"

  result = yield(job)
  @success_count += 1
  result
rescue StandardError => e
  log_debug "Bypassed execution failed: #{e.class}"
  raise
end

#execute_with_breaker(job) {|Job| ... } ⇒ Object

Execute a job with circuit breaker protection.

Parameters:

  • job (Job) —

    The job to execute

Yields:

  • (Job) —

    Block that executes the job

Returns:

  • (Object) —

    The execution result

Raises:



43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 43

def execute_with_breaker(job, &)
  @execution_count += 1

  log_debug "Executing job '#{job.name}' with circuit breaker protection"

  check_and_call_breaker(job, &)
rescue CircuitOpenError => e
  @blocked_count += 1
  log_debug "Job '#{job.name}' blocked by circuit breaker: #{e.message}"
  raise
rescue StandardError => e
  log_debug "Job '#{job.name}' failed with #{e.class}"
  raise
end

#failure_count ⇒ Integer

Get the failure count.

Returns:

  • (Integer) —

    Number of failures recorded



89
90
91
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 89

def failure_count
  @breaker.failure_count
end

#half_open? ⇒ Boolean

Check if the circuit is currently half-open.

Returns:

  • (Boolean) —

    true if circuit is half-open



75
76
77
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 75

def half_open?
  @breaker.half_open?
end

#last_failure_time ⇒ Time?

Get the last failure time.

Returns:

  • (Time, nil) —

    Last failure time or nil



96
97
98
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 96

def last_failure_time
  @breaker.last_failure_time
end

#open? ⇒ Boolean

Check if the circuit is currently open.

Returns:

  • (Boolean) —

    true if circuit is open



61
62
63
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 61

def open?
  @breaker.open?
end

#open_circuit! ⇒ Object

Manually open the circuit (for testing or emergency).



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

def open_circuit!
  @breaker.instance_variable_get(:@mutex).synchronize do
    @breaker.instance_variable_set(:@state, CircuitBreaker::STATE_OPEN)
    @breaker.instance_variable_set(:@failure_count, @breaker.threshold)
    @breaker.instance_variable_set(:@last_failure_time, Time.now)
  end
  log_debug "Circuit manually opened for job '#{@job_name}'"
end

#reset! ⇒ Object

Reset the circuit breaker to closed state.



101
102
103
104
105
106
107
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 101

def reset!
  @breaker.reset
  @execution_count = 0
  @success_count = 0
  @blocked_count = 0
  log_debug "Circuit breaker reset for job '#{@job_name}'"
end

#state ⇒ Symbol

Get the current circuit breaker state.

Returns:

  • (Symbol) —

    The state (:closed, :open, :half_open)



82
83
84
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 82

def state
  @breaker.state
end

#state_description ⇒ String

Get the current state as a human-readable string.

Returns:

  • (String) —

    State description



123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 123

def state_description
  case state
  when CircuitBreaker::STATE_CLOSED
    "CLOSED (normal operation)"
  when CircuitBreaker::STATE_OPEN
    "OPEN (blocking requests, #{failure_count}/#{@breaker.threshold} failures)"
  when CircuitBreaker::STATE_HALF_OPEN
    "HALF_OPEN (testing recovery, #{@breaker.instance_variable_get(:@success_count)}/#{@breaker.half_open_calls} successes)"
  else
    "UNKNOWN"
  end
end

#stats ⇒ Hash

Get circuit breaker statistics including orchestrator metrics.

Returns:

  • (Hash) —

    Statistics and metrics



112
113
114
115
116
117
118
 
# File 'lib/fractor/workflow/circuit_breaker_orchestrator.rb', line 112

def stats
  @breaker.stats.merge(
    execution_count: @execution_count,
    success_count: @success_count,
    blocked_count: @blocked_count,
  )
end