Class: Pgbus::Client::ConnectionHealth

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/client/connection_health.rb

Overview

In-memory, process-local circuit breaker for database-down conditions.

Distinct from Pgbus::CircuitBreaker (which is per-queue and persists its pause state in the database — useless when the database itself is down). This latch is a single shared object owned by Pgbus::Client that trips on raw connection failures across any operation, so a fleet of workers stops hammering a dead database and flooding the error tracker for the whole outage (issue #197).

State machine:

closed    -> open       after OPEN_THRESHOLD consecutive ConnectionErrors
open      -> half_open  once the backoff window elapses (admits ONE probe)
half_open -> closed     on a successful probe (resets backoff)
half_open -> open       on a failed probe (backoff doubles, capped)

Thresholds are constants, not configuration, mirroring CircuitBreaker: the values rarely need tuning and exposing them never proved useful.

Thread safety: a single Mutex serializes every state transition. The guarded operation itself runs outside the lock (only the gate decision and the outcome recording are inside), so a slow probe never blocks other worker threads from failing fast.

Constant Summary collapse

OPEN_THRESHOLD =

Consecutive ConnectionErrors that trip the breaker from closed to open.

5
BASE_BACKOFF =

Initial backoff (seconds) on the first trip. Doubles on each re-open.

1.0
MAX_BACKOFF =

Cap on the exponential backoff (seconds). After ~6 re-opens the curve plateaus here so a perpetually-down database stops probing more than once a minute.

60.0

Instance Method Summary collapse

Constructor Details

#initialize(clock: -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) }, on_open: nil, on_close: nil) ⇒ ConnectionHealth

Returns a new instance of ConnectionHealth.



39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/pgbus/client/connection_health.rb', line 39

def initialize(clock: -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) },
               on_open: nil, on_close: nil)
  @clock = clock
  @on_open = on_open
  @on_close = on_close
  @mutex = Mutex.new
  @state = :closed
  @failure_count = 0
  @trip_count = 0
  @open_until = nil
end

Instance Method Details

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


76
77
78
 
# File 'lib/pgbus/client/connection_health.rb', line 76

def closed?
  @mutex.synchronize { @state == :closed }
end

#open? ⇒ Boolean

Returns:

  • (Boolean) 


80
81
82
 
# File 'lib/pgbus/client/connection_health.rb', line 80

def open?
  @mutex.synchronize { @state == :open }
end

#run_guarded ⇒ Object

Gate an operation through the breaker.

  • closed / half-open-probe-admitted: yields, records the outcome.
  • open (window not elapsed) or half-open (probe already in flight): raises Pgbus::ConnectionCircuitOpenError without yielding — no pool checkout, no ConnectionError, no ErrorReporter noise.

The breaker never swallows an error — every exception propagates. A ConnectionError in the closed state counts toward the trip threshold; a failure of the single half-open probe (on ANY error) re-opens the breaker; any other closed-state error is not counted.



62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/pgbus/client/connection_health.rb', line 62

def run_guarded
  admitted = admit_or_raise # :run (was closed) or :probe (was open)

  begin
    result = yield
  rescue StandardError => e
    record_failure(admitted, e)
    raise
  end

  record_success(admitted)
  result
end