Class: Pgbus::Process::Worker

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

Constant Summary collapse

NOTIFY_FALLBACK_POLL_SECONDS = 
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. Rather than downgrade to blind polling until process restart, the loop retries from ensure_notify_listener on an exponential backoff: NOTIFY_RETRY_BASE doubling up to NOTIFY_RETRY_MAX. Constant-tuned, matching the CircuitBreaker/Dispatcher precedent.

5
NOTIFY_RETRY_MAX_SECONDS = 
300
WILDCARD_REFRESH_INTERVAL =

seconds

30
MISSING_QUEUE_REGEX =

Matches the physical queue name inside a "relation "pgmq.q_foo" does not exist" error. Frozen module constant to avoid recompiling the regex on every queue-missing error in hot fetch/read paths.

/pgmq\.q_(\w+)/
RESTORE_COOLDOWN_BASE =

Exponential backoff bounds for restoring evicted queues (issue #209). After the worker's queues are fully evicted (queue table permanently deleted), the first restore waits RESTORE_COOLDOWN_BASE seconds; each consecutive failed restore doubles the wait up to RESTORE_COOLDOWN_MAX. A successful fetch resets the streak so a recreated queue restores promptly. Constant-tuned, matching the NOTIFY_RETRY precedent.

30
RESTORE_COOLDOWN_MAX = 
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(queues:, threads: 5, config: Pgbus.configuration, single_active_consumer: false, consumer_priority: 0, execution_mode: :threads, group_mode: nil, liveness_pipe: nil, rate_counter: nil, wake_signal: nil, stat_buffer: :default, notify_listener: nil, notify_retry_at: 0.0, notify_retry_backoff: NOTIFY_RETRY_BASE_SECONDS, started_at_monotonic: nil) ⇒ Worker

The collaborators below (rate_counter, wake_signal, stat_buffer) and the recycle clock (started_at_monotonic) accept injected seeds so tests can observe or stub them without poking private ivars. All default to the exact values production constructs, so behavior is unchanged.



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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
100
101
 
# File 'lib/pgbus/process/worker.rb', line 27

def initialize(queues:, threads: 5, config: Pgbus.configuration,
               single_active_consumer: false, consumer_priority: 0,
               execution_mode: :threads, group_mode: nil, liveness_pipe: nil,
               rate_counter: nil, wake_signal: nil, stat_buffer: :default,
               notify_listener: nil, notify_retry_at: 0.0,
               notify_retry_backoff: NOTIFY_RETRY_BASE_SECONDS,
               started_at_monotonic: nil)
  @queues = Array(queues)
  @initial_queues = @queues.dup.freeze
  @wildcard = @queues.include?("*")
  @threads = threads
  @config = config
  @execution_mode = ExecutionPools.normalize_mode(execution_mode)
  @group_mode = case group_mode
                when nil then nil
                when Symbol then group_mode
                when String then group_mode.to_sym
                else
                  raise ArgumentError,
                        "Invalid group_mode type: #{group_mode.class}. Must be nil, String, or Symbol"
                end
  unless Pgbus::Configuration::VALID_GROUP_MODES.include?(@group_mode)
    raise ArgumentError,
          "Invalid group_mode: #{@group_mode.inspect}. Must be nil, :fifo, or :round_robin"
  end
  @single_active_consumer = single_active_consumer
  @consumer_priority = consumer_priority
  @lifecycle = Lifecycle.new
  @last_wildcard_resolve = nil
  @jobs_processed = Concurrent::AtomicFixnum.new(0)
  @jobs_failed = Concurrent::AtomicFixnum.new(0)
  @in_flight = Concurrent::AtomicFixnum.new(0)
  @loop_tick_at = Concurrent::AtomicReference.new(nil)
  @rate_counter = rate_counter || RateCounter.new(:processed, :failed, :dequeued)
  @started_at = Time.current
  @started_at_monotonic = started_at_monotonic || monotonic_now
  # 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
  @executor = Pgbus::ActiveJob::Executor.new(stat_buffer: @stat_buffer)
  @wake_signal = wake_signal || WakeSignal.new
  @pool = ExecutionPools.build(
    mode: @execution_mode,
    capacity: threads,
    on_state_change: -> { @wake_signal.notify! }
  )
  @circuit_breaker = Pgbus::CircuitBreaker.new(config: config)
  @drain_started_at = nil
  # Evict/restore cooldown state (issue #209). When a permanently-deleted
  # queue is evicted down to an empty list, restoring the initial queues
  # immediately re-triggers the same undefined-table error every loop tick.
  # These back off the restore attempt on an exponential schedule instead.
  @restore_streak = 0
  @last_evicted_at = nil
  @deferral_warned = false
  @queue_lock = QueueLock.new if @single_active_consumer
  @notify_listener = notify_listener
  @notify_retry_at = notify_retry_at
  @notify_retry_backoff = notify_retry_backoff
  # OS-level liveness channel to the supervisor watchdog. Optional: nil
  # unless the supervisor forked us with one. Written from stamp_loop_tick
  # so the watchdog can detect a wedged worker even when the database (and
  # thus the Heartbeat's loop_tick_at) is unavailable.
  @liveness_pipe = liveness_pipe
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



10
11
12
 
# File 'lib/pgbus/process/worker.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/worker.rb', line 10

def execution_mode
  @execution_mode
end

#lifecycle ⇒ Object (readonly)

Returns the value of attribute lifecycle.



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

def lifecycle
  @lifecycle
end

#notify_listener ⇒ Object

notify_listener / notify_retry_at / notify_retry_backoff are writable so tests can seed the self-healing listener state, re-arm the backoff window between calls, and simulate start_notify_listener assigning the listener from inside a stub (production mutates all three in the run loop).



21
22
23
 
# File 'lib/pgbus/process/worker.rb', line 21

def notify_listener
  @notify_listener
end

#notify_retry_at ⇒ Object

notify_listener / notify_retry_at / notify_retry_backoff are writable so tests can seed the self-healing listener state, re-arm the backoff window between calls, and simulate start_notify_listener assigning the listener from inside a stub (production mutates all three in the run loop).



21
22
23
 
# File 'lib/pgbus/process/worker.rb', line 21

def notify_retry_at
  @notify_retry_at
end

#notify_retry_backoff ⇒ Object

notify_listener / notify_retry_at / notify_retry_backoff are writable so tests can seed the self-healing listener state, re-arm the backoff window between calls, and simulate start_notify_listener assigning the listener from inside a stub (production mutates all three in the run loop).



21
22
23
 
# File 'lib/pgbus/process/worker.rb', line 21

def notify_retry_backoff
  @notify_retry_backoff
end

#queues ⇒ Object (readonly)

Returns the value of attribute queues.



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

def queues
  @queues
end

#rate_counter ⇒ Object (readonly)

Returns the value of attribute rate_counter.



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

def rate_counter
  @rate_counter
end

#restore_streak ⇒ Object (readonly)

Returns the value of attribute restore_streak.



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

def restore_streak
  @restore_streak
end

#stat_buffer ⇒ Object

stat_buffer is writable so a test can swap in a buffer double after construction to assert graceful_shutdown / check_recycle flush it. The executor captured the original buffer at construction, but these paths flush @stat_buffer directly, so the swap is observable.



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

def stat_buffer
  @stat_buffer
end

#threads ⇒ Object (readonly)

Returns the value of attribute threads.



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

def threads
  @threads
end

#wake_signal ⇒ Object (readonly)

Returns the value of attribute wake_signal.



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

def wake_signal
  @wake_signal
end

Instance Method Details

#graceful_shutdown ⇒ Object 



182
183
184
185
186
187
188
189
190
191
192
193
194
 
# File 'lib/pgbus/process/worker.rb', line 182

def graceful_shutdown
  Pgbus.logger.info { "[Pgbus] Worker shutting down gracefully..." }
  Pgbus.stopping = true
  @lifecycle.transition_to(:draining)
  # Flush buffered stats at drain entry so the ≤ stats_flush_interval /
  # stats_flush_size window isn't lost if the supervisor watchdog SIGKILLs
  # a stalled worker before the drain-loop shutdown flush runs. Runs on the
  # main loop thread (signals are dispatched via process_signals, not in
  # trap context), so the DB write is safe. flush is thread-safe and
  # no-ops when the buffer is empty.
  @stat_buffer&.flush
  @wake_signal.notify!
end

#immediate_shutdown ⇒ Object 



196
197
198
199
200
201
202
 
# File 'lib/pgbus/process/worker.rb', line 196

def immediate_shutdown
  Pgbus.logger.warn { "[Pgbus] Worker shutting down immediately!" }
  Pgbus.stopping = true
  @lifecycle.transition_to!(:stopped)
  @wake_signal.notify!
  @pool.kill
end

#in_flight=(count) ⇒ Object 



126
127
128
 
# File 'lib/pgbus/process/worker.rb', line 126

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

#jobs_processed=(count) ⇒ Object

Test seams for the atomic counters recycle logic and prefetch flow-control consult. Production only increments these during message handling; seeding them lets a test cross a threshold without running thousands of real jobs.



122
123
124
 
# File 'lib/pgbus/process/worker.rb', line 122

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 — NOT monotonic — so it stays comparable across the process boundary the supervisor watchdog reads it over. nil until the first stamp_loop_tick.



134
135
136
 
# File 'lib/pgbus/process/worker.rb', line 134

def last_loop_tick
  @loop_tick_at.get
end

#run ⇒ Object 



149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/pgbus/process/worker.rb', line 149

def run
  setup_signals
  start_heartbeat
  resolve_wildcard_queues
  start_notify_listener
  @lifecycle.transition_to!(:running)
  Pgbus.logger.info do
    "[Pgbus] Worker started: queues=#{queues.join(",")} threads=#{threads} " \
      "mode=#{@execution_mode} notify_wakeup=#{notify_wakeup?} pid=#{::Process.pid}"
  end

  loop do
    stamp_loop_tick
    process_signals
    check_recycle
    refresh_wildcard_queues
    ensure_notify_listener

    break if @lifecycle.stopped?
    # quiesced? (all slots free), not idle? (any slot free) — exiting
    # with work still in flight abandons those jobs to the 30s
    # wait_for_termination timeout in shutdown. Bounded by
    # config.drain_timeout so a stuck job can't wedge the loop forever.
    break if @lifecycle.draining? && (@pool.quiesced? || drain_deadline_exceeded?)

    claim_and_execute if @lifecycle.can_process?
    @stat_buffer&.flush_if_due
    @wake_signal.wait(timeout: config.polling_interval) if @lifecycle.draining? || @lifecycle.paused?
  end

  shutdown
end

#stats ⇒ Object 



103
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/pgbus/process/worker.rb', line 103

def stats
  {
    jobs_processed: @jobs_processed.value,
    jobs_failed: @jobs_failed.value,
    in_flight: @in_flight.value,
    state: @lifecycle.state,
    execution_mode: @execution_mode,
    consumer_priority: @consumer_priority,
    single_active_consumer: @single_active_consumer,
    locked_queues: @queue_lock&.held_queues || [],
    rates: @rate_counter.rates,
    started_at: @started_at
  }.merge(@pool.metadata)
end