Class: Pgbus::Process::Consumer

Inherits:
Object
  • Object
show all
Includes:
SignalHandler
Defined in:
lib/pgbus/process/consumer.rb

Constant Summary collapse

NOTIFY_FALLBACK_POLL_SECONDS =

Mirrors Worker::NOTIFY_FALLBACK_POLL_SECONDS. When a live NotifyListener drives wake-up latency, the empty-read wait is a safety net rather than the steady-state cadence, so it rises to this ceiling: one missed NOTIFY costs bounded latency, never a stuck queue.

15
NOTIFY_RETRY_BASE_SECONDS =

NotifyListener startup can fail on a transient boot-time condition (DB restarting, failover, DNS blip) or its thread can die mid-run. The loop retries from ensure_notify_listener on an exponential backoff: NOTIFY_RETRY_BASE doubling up to NOTIFY_RETRY_MAX. Matches Worker.

5
NOTIFY_RETRY_MAX_SECONDS = 
300

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from SignalHandler

#handled_signals, included, #interruptible_sleep, #process_signals, #restore_signals, #setup_signals

Constructor Details

#initialize(topics:, threads: 3, config: Pgbus.configuration, execution_mode: :threads, queue_names: nil, liveness_pipe: nil, stat_buffer: :default, notify_listener: nil, notify_retry_at: 0.0, notify_retry_backoff: NOTIFY_RETRY_BASE_SECONDS, started_at_monotonic: nil) ⇒ Consumer

The notify-listener lifecycle state (notify_listener, notify_retry_at, notify_retry_backoff), the recycle clock (started_at_monotonic), and the resolved subscription set (queue_names) accept injected seeds so tests can drive the run loop from a known state without poking private ivars. All default to the values production initializes to; queue_names defaults to nil, meaning "derive from the registry in setup_subscriptions".



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/pgbus/process/consumer.rb', line 56

def initialize(topics:, threads: 3, config: Pgbus.configuration, execution_mode: :threads,
               queue_names: nil, liveness_pipe: nil, stat_buffer: :default,
               notify_listener: nil, notify_retry_at: 0.0,
               notify_retry_backoff: NOTIFY_RETRY_BASE_SECONDS,
               started_at_monotonic: nil)
  @topics = Array(topics)
  @threads = threads
  @config = config
  @execution_mode = ExecutionPools.normalize_mode(execution_mode)
  @shutting_down = false
  @recycling = false
  @jobs_processed = Concurrent::AtomicFixnum.new(0)
  @loop_tick_at = Concurrent::AtomicReference.new(nil)
  @started_at_monotonic = started_at_monotonic || monotonic_now
  @wake_signal = WakeSignal.new
  @pool = ExecutionPools.build(
    mode: @execution_mode,
    capacity: threads,
    on_state_change: -> { @wake_signal.notify! }
  )
  @registry = EventBus::Registry.instance
  @circuit_breaker = Pgbus::CircuitBreaker.new(config: config)
  # stat_buffer: :default means "build one iff config.stats_enabled";
  # passing an explicit value (including nil) overrides that for tests.
  @stat_buffer =
    if stat_buffer == :default
      if config.stats_enabled
        Pgbus::StatBuffer.new(
          flush_size: config.stats_flush_size,
          flush_interval: config.stats_flush_interval
        )
      end
    else
      stat_buffer
    end
  @queue_names = queue_names
  @notify_listener = notify_listener
  @notify_retry_at = notify_retry_at
  @notify_retry_backoff = notify_retry_backoff
  # OS-level liveness channel to the supervisor watchdog. nil unless the
  # supervisor forked us with one. Written from stamp_loop_tick so the
  # watchdog can detect a wedged consumer even when the database is down.
  @liveness_pipe = liveness_pipe
end

Instance Attribute Details

#circuit_breaker ⇒ Object (readonly)

Returns the value of attribute circuit_breaker.



10
11
12
 
# File 'lib/pgbus/process/consumer.rb', line 10

def circuit_breaker
  @circuit_breaker
end

#config ⇒ Object (readonly)

Returns the value of attribute config.



10
11
12
 
# File 'lib/pgbus/process/consumer.rb', line 10

def config
  @config
end

#execution_mode ⇒ Object (readonly)

Returns the value of attribute execution_mode.



10
11
12
 
# File 'lib/pgbus/process/consumer.rb', line 10

def execution_mode
  @execution_mode
end

#notify_listener ⇒ Object

notify_listener is writable so tests can simulate a start_notify_listener success from inside a stub (production sets it in start_notify_listener). notify_retry_at is writable so a test can re-arm the backoff window between successive ensure_notify_listener calls.



16
17
18
 
# File 'lib/pgbus/process/consumer.rb', line 16

def notify_listener
  @notify_listener
end

#notify_retry_at ⇒ Object

notify_listener is writable so tests can simulate a start_notify_listener success from inside a stub (production sets it in start_notify_listener). notify_retry_at is writable so a test can re-arm the backoff window between successive ensure_notify_listener calls.



16
17
18
 
# File 'lib/pgbus/process/consumer.rb', line 16

def notify_retry_at
  @notify_retry_at
end

#notify_retry_backoff ⇒ Object (readonly)

Returns the value of attribute notify_retry_backoff.



10
11
12
 
# File 'lib/pgbus/process/consumer.rb', line 10

def notify_retry_backoff
  @notify_retry_backoff
end

#queue_names ⇒ Object (readonly)

Returns the value of attribute queue_names.



10
11
12
 
# File 'lib/pgbus/process/consumer.rb', line 10

def queue_names
  @queue_names
end

#stat_buffer ⇒ Object

stat_buffer is writable so a test can swap in a buffer double after construction and assert graceful_shutdown / check_recycle / shutdown flush it (mirrors Worker#stat_buffer).



20
21
22
 
# File 'lib/pgbus/process/consumer.rb', line 20

def stat_buffer
  @stat_buffer
end

#threads ⇒ Object (readonly)

Returns the value of attribute threads.



10
11
12
 
# File 'lib/pgbus/process/consumer.rb', line 10

def threads
  @threads
end

#topics ⇒ Object (readonly)

Returns the value of attribute topics.



10
11
12
 
# File 'lib/pgbus/process/consumer.rb', line 10

def topics
  @topics
end

#wake_signal ⇒ Object (readonly)

Returns the value of attribute wake_signal.



10
11
12
 
# File 'lib/pgbus/process/consumer.rb', line 10

def wake_signal
  @wake_signal
end

Instance Method Details

#graceful_shutdown ⇒ Object 



134
135
136
137
138
139
140
141
142
 
# File 'lib/pgbus/process/consumer.rb', line 134

def graceful_shutdown
  @shutting_down = true
  # Flush buffered stats at drain entry so the window since the last flush
  # isn't lost if the supervisor watchdog SIGKILLs a stalled consumer
  # before shutdown runs. Same-thread (signals dispatched via
  # process_signals, not trap context), so the DB write is safe.
  @stat_buffer&.flush
  @wake_signal.notify!
end

#immediate_shutdown ⇒ Object 



144
145
146
147
148
 
# File 'lib/pgbus/process/consumer.rb', line 144

def immediate_shutdown
  @shutting_down = true
  @wake_signal.notify!
  @pool.kill
end

#jobs_processed ⇒ Object 



26
27
28
 
# File 'lib/pgbus/process/consumer.rb', line 26

def jobs_processed
  @jobs_processed.value
end

#jobs_processed=(count) ⇒ Object

Seed the processed-job counter. Used by tests to drive the recycle thresholds without running thousands of real jobs; production only ever increments it via the AtomicFixnum during message handling.



33
34
35
 
# File 'lib/pgbus/process/consumer.rb', line 33

def jobs_processed=(count)
  @jobs_processed.value = count
end

#last_loop_tick ⇒ Object

The last wall-clock loop-tick stamp (Time.now.to_f) fed to the heartbeat's loop_tick_supplier. Wall-clock so it stays comparable across the process boundary the supervisor watchdog reads it over. nil until the first stamp_loop_tick.



105
106
107
 
# File 'lib/pgbus/process/consumer.rb', line 105

def last_loop_tick
  @loop_tick_at.get
end

#run ⇒ Object 



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/pgbus/process/consumer.rb', line 109

def run
  setup_signals
  start_heartbeat
  setup_subscriptions
  start_notify_listener
  Pgbus.logger.info do
    "[Pgbus] Consumer started: topics=#{topics.join(",")} threads=#{threads} " \
      "notify_wakeup=#{notify_wakeup?} pid=#{::Process.pid}"
  end

  loop do
    stamp_loop_tick
    process_signals
    check_recycle
    ensure_notify_listener

    break if @shutting_down

    consume
    @stat_buffer&.flush_if_due
  end

  shutdown
end

#shutting_down? ⇒ Boolean

Returns:

  • (Boolean) 


22
23
24
 
# File 'lib/pgbus/process/consumer.rb', line 22

def shutting_down?
  @shutting_down
end