Class: Pgbus::Client

Inherits:
Object
  • Object
show all
Includes:
EnsureStreamQueue, NotifyStream, ReadAfter
Defined in:
lib/pgbus/client.rb,
lib/pgbus/client/read_after.rb,
lib/pgbus/client/notify_stream.rb,
lib/pgbus/client/resizable_pool.rb,
lib/pgbus/client/connection_health.rb,
lib/pgbus/client/ensure_stream_queue.rb

Defined Under Namespace

Modules: EnsureStreamQueue, NotifyStream, ReadAfter Classes: ConnectionHealth, ResizablePool, WedgedReadTimeout

Constant Summary collapse

PGMQ_INSTALL_LOCK_KEY =

Fixed advisory-lock key serializing pgmq schema installation across processes (issue #397). "pgmqinst" in ASCII hex — arbitrary but stable; it only has to be identical in every process that can install.

0x70676D71_696E7374
NOTIFY_THROTTLE_MS =

Throttle window for PGMQ's enable_notify_insert trigger. Postgres NOTIFYs are coalesced into one wake-up per window, so a value of 250ms means: at most 4 broadcasts/sec per queue, regardless of insert rate. The trigger is a Postgres-level concern; exposing it as a setting never came up in practice and changing it on the fly would require re-running the trigger DDL on every queue.

250

Constants included from NotifyStream

NotifyStream::NOTIFY_PAYLOAD_LIMIT_BYTES

Constants included from ReadAfter

ReadAfter::DEFAULT_LIMIT

Class Attribute Summary collapse

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from NotifyStream

#notify_stream

Methods included from EnsureStreamQueue

#ensure_stream_queue

Methods included from ReadAfter

#read_after, #stream_current_msg_id, #stream_oldest_msg_id

Constructor Details

#initialize(config = Pgbus.configuration, schema_ensured: false) ⇒ Client

schema_ensured: lets a caller (in practice, tests) skip the one-time PGMQ schema install probe by asserting the schema already exists. Defaults to false so production always runs the check on first queue access.



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
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
 
# File 'lib/pgbus/client.rb', line 85

def initialize(config = Pgbus.configuration, schema_ensured: false)
  self.class.load_pgmq_gem!
  @config = config
  conn_opts = config.connection_options
  @shared_connection = conn_opts.is_a?(Proc)

  if @shared_connection
    # When using the Rails lambda path (-> { AR::Base.connection.raw_connection }),
    # the Proc returns the same underlying PG::Connection that ActiveRecord uses.
    # PG::Connection (libpq) is not thread-safe — concurrent access causes
    # segfaults and result corruption. Force pool_size=1 and serialize all
    # operations through a mutex.
    @pgmq = PGMQ::Client.new(conn_opts, pool_size: 1, pool_timeout: config.pool_timeout)
    @pgmq_mutex = Mutex.new
    # No dedicated streams pool on this path: a second PGMQ::Client would
    # still funnel through the same non-thread-safe AR raw_connection.
    # Stream publish + replay share the single serialized connection —
    # @streams_pgmq aliases @pgmq so the code paths are uniform, and
    # #with_streams_connection falls back to with_raw_connection.
    @streams_pgmq = @pgmq
  else
    # With a String URL or Hash params, pgmq-ruby creates its own dedicated
    # PG::Connection per pool slot — no shared state with ActiveRecord.
    # Use the resolved pool size (auto-tuned from worker thread counts
    # unless explicitly set) and let pgmq-ruby's connection_pool handle
    # concurrency internally (no mutex needed).
    #
    # Bound reads with libpq-native mechanisms baked into the connection
    # options (issue #198): a server-side statement_timeout for a slow query,
    # plus client-side tcp_user_timeout + keepalives for a dead/hung socket.
    # Both raise clean PG errors — no Ruby Timeout, no Thread#raise. Only
    # safe on this dedicated-connection branch — never on the shared-AR Proc
    # path, where statement_timeout would leak into application queries.
    conn_opts = wrap_session_gucs(apply_connection_bounds(conn_opts))
    @pgmq = PGMQ::Client.new(conn_opts, pool_size: config.resolved_pool_size, pool_timeout: config.pool_timeout)
    @pgmq_mutex = nil
    # Dedicated streams pool (issue #315): isolates the durable-stream
    # publish INSERT (#send_stream_message) and the dispatcher's per-wake
    # replay reads (#read_after) from the job pool, so a saturated worker
    # pool can't delay a broadcast on pool checkout, and each wake reuses a
    # persistent connection instead of a fresh PG.connect per call. Its own
    # PGMQ::Client → its own connection_pool, sized independently of worker
    # thread counts.
    # Build the streams pool from streams_pool_connection_options (defaults
    # to streams_connection_options so a separate streams DB carries the
    # pool with it — issue #315 — but overridable via streams_pool_* so a
    # pooler-bypass install keeps the pool off the direct port — issue
    # #358), bounds-applied, and tagged with a per-process application_name
    # so the autoscaler can count peer processes from pg_stat_activity
    # (issue #323 P1/P2). Snapshot it so a hot-swap rebuilds a
    # byte-identical pool at a new size.
    @streams_conn_opts = wrap_session_gucs(
      tag_application_name(
        apply_connection_bounds(config.streams_pool_connection_options)
      )
    )
    @streams_pgmq = PGMQ::Client.new(@streams_conn_opts, pool_size: config.streams_pool_size,
                                                         pool_timeout: config.streams_pool_timeout)
  end

  # Wrap the streams pool so its live reference can be atomically hot-swapped
  # to a new size under load without losing broadcasts or leaking connections
  # (issue #323 spike; #resize_streams_pool). All streams-pool access goes
  # through this — see #streams_pool. Default behavior with no swap is
  # byte-identical (one AtomicReference read + a counter bump per op).
  @streams_pool = ResizablePool.new(
    @streams_pgmq,
    shared: @shared_connection,
    drain_timeout: config.streams_pool_timeout + 1.0,
    logger: Pgbus.logger
  )

  @queues_created = Concurrent::Map.new
  @stream_indexes_created = Concurrent::Map.new
  # Guards the one-time build of the publisher autoscale trigger (issue #323).
  # NOT @pgmq_mutex — that is nil on the dedicated path (the only path the
  # trigger exists on), so it wouldn't serialize concurrent first-publishers.
  @streams_trigger_mutex = Mutex.new
  @queue_strategy = QueueFactory.for(config)
  @schema_ensured = schema_ensured
  @connection_health = ConnectionHealth.new(
    on_open: method(:log_circuit_open),
    on_close: method(:log_circuit_close)
  )
  # Snapshot whether libpq's baked-in read bounds fully cover a hung socket
  # on this host/connection, so the read path can skip the Ruby Timeout
  # last resort. Computed once: @shared_connection, config.read_timeout
  # (which apply_connection_bounds also snapshots), the platform, and the
  # linked libpq version are all fixed for a Client's lifetime.
  @libpq_read_bounds_effective = libpq_read_bounds_effective?
  warn_shared_connection_read_bounds
end

Class Attribute Details

.pgmq_install_mutex ⇒ Object (readonly)

Returns the value of attribute pgmq_install_mutex.



56
57
58
 
# File 'lib/pgbus/client.rb', line 56

def pgmq_install_mutex
  @pgmq_install_mutex
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



18
19
20
 
# File 'lib/pgbus/client.rb', line 18

def config
  @config
end

#connection_health ⇒ Object (readonly)

Returns the value of attribute connection_health.



18
19
20
 
# File 'lib/pgbus/client.rb', line 18

def connection_health
  @connection_health
end

#pgmq ⇒ Object (readonly)

Returns the value of attribute pgmq.



18
19
20
 
# File 'lib/pgbus/client.rb', line 18

def pgmq
  @pgmq
end

Class Method Details

.load_pgmq_gem! ⇒ Object

Load the pgmq-ruby gem, defining the PGMQ module before requiring it so Zeitwerk's eager_load (called inside pgmq.rb) can resolve the constant. Without the pre-definition, Ruby 4.0 + Zeitwerk 2.7.5 raises NameError because eager_load runs const_get(:Client) on PGMQ before the module is defined. Extracted as a class method so unit specs that fake PGMQ::Client can stub this (a per-example class-method stub, torn down cleanly) instead of the global Kernel#require, which — if stubbed before pgmq is genuinely loaded — permanently prevents the real gem from ever loading.



75
76
77
78
79
80
 
# File 'lib/pgbus/client.rb', line 75

def self.load_pgmq_gem!
  PGMQ_REQUIRE_MUTEX.synchronize do
    Object.const_set(:PGMQ, Module.new) unless defined?(::PGMQ)
    require "pgmq"
  end
end

Instance Method Details

#archive_batch(queue_name, msg_ids, prefixed: true) ⇒ Object

Batch archive — moves multiple messages to the archive table in one call.



520
521
522
523
524
525
 
# File 'lib/pgbus/client.rb', line 520

def archive_batch(queue_name, msg_ids, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.archive_batch(name, msg_ids) }
  end
end

#archive_message(queue_name, msg_id, prefixed: true) ⇒ Object

Archive a message. Pass prefixed: false when queue_name is already the full PGMQ queue name.



512
513
514
515
516
517
 
# File 'lib/pgbus/client.rb', line 512

def archive_message(queue_name, msg_id, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.archive(name, msg_id) }
  end
end

#bind_topic(pattern, queue_name) ⇒ Object

Topic routing



815
816
817
818
819
820
821
 
# File 'lib/pgbus/client.rb', line 815

def bind_topic(pattern, queue_name)
  full_name = config.queue_name(queue_name)
  with_stale_connection_retry do
    ensure_queue(queue_name)
    synchronized { @pgmq.bind_topic(pattern, full_name) }
  end
end

#close ⇒ Object 



836
837
838
839
840
841
842
843
844
845
846
847
848
849
 
# File 'lib/pgbus/client.rb', line 836

def close
  # Stop the publisher autoscale executor (if one was ever built) so its
  # background thread doesn't leak (issue #323). Outside `synchronized` — it
  # takes no pool lock and shutdown waits on a possibly-running check.
  @streams_pool_trigger.shutdown if defined?(@streams_pool_trigger) && @streams_pool_trigger
  synchronized do
    @pgmq.close
    # Close the CURRENT streams pool too (issue #315) so its connections
    # don't leak. close_current reads the live (possibly hot-swapped, #323)
    # pool once under the swap mutex and skips it when it aliases @pgmq (the
    # shared-AR path) so we don't double-close the same pool.
    @streams_pool.close_current(job_pool: @pgmq)
  end
end

#configured_queues ⇒ Object

The logical queue names pgbus expects to exist based on the configuration (default queue + worker capsules + recurring tasks). Public wrapper around collect_configured_queues so the doctor can diff configured-vs-existing queues without reaching into PGMQ or config internals directly.



253
254
255
 
# File 'lib/pgbus/client.rb', line 253

def configured_queues
  collect_configured_queues
end

#convert_archive_partitioned(queue_name, partition_interval: "10000", retention_interval: "100000", leading_partition: 10) ⇒ Object

--- Archive partitioning (requires pg_partman extension) ---



799
800
801
802
803
804
805
806
807
808
809
810
811
812
 
# File 'lib/pgbus/client.rb', line 799

def convert_archive_partitioned(queue_name, partition_interval: "10000", retention_interval: "100000",
                                leading_partition: 10)
  full_name = config.queue_name(queue_name)
  with_stale_connection_retry do
    synchronized do
      @pgmq.convert_archive_partitioned(
        full_name,
        partition_interval: partition_interval,
        retention_interval: retention_interval,
        leading_partition: leading_partition
      )
    end
  end
end

#create_fifo_index(queue_name) ⇒ Object

--- FIFO index management (PGMQ v1.11.0+) ---



762
763
764
765
766
767
 
# File 'lib/pgbus/client.rb', line 762

def create_fifo_index(queue_name)
  full_name = config.queue_name(queue_name)
  with_stale_connection_retry do
    synchronized { @pgmq.create_fifo_index(full_name) }
  end
end

#create_fifo_indexes_all ⇒ Object 



769
770
771
772
773
 
# File 'lib/pgbus/client.rb', line 769

def create_fifo_indexes_all
  with_stale_connection_retry do
    synchronized { @pgmq.create_fifo_indexes_all }
  end
end

#delete_batch(queue_name, msg_ids, prefixed: true) ⇒ Object

Batch delete — permanently removes multiple messages in one call.



528
529
530
531
532
533
 
# File 'lib/pgbus/client.rb', line 528

def delete_batch(queue_name, msg_ids, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.delete_batch(name, msg_ids) }
  end
end

#delete_message(queue_name, msg_id, prefixed: true) ⇒ Object

Delete a message. Pass prefixed: false when queue_name is already the full PGMQ queue name (e.g. from priority sub-queues or dashboard).



503
504
505
506
507
508
 
# File 'lib/pgbus/client.rb', line 503

def delete_message(queue_name, msg_id, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.delete(name, msg_id) }
  end
end

#drop_queue(queue_name, prefixed: true) ⇒ Object 



654
655
656
657
658
659
660
661
 
# File 'lib/pgbus/client.rb', line 654

def drop_queue(queue_name, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  result = with_stale_connection_retry do
    synchronized { @pgmq.drop_queue(name) }
  end
  @queues_created.delete(name)
  result
end

#ensure_all_queues ⇒ Object 



316
317
318
319
320
 
# File 'lib/pgbus/client.rb', line 316

def ensure_all_queues
  queue_names = collect_configured_queues
  Pgbus.logger.info { "[Pgbus] Bootstrapping #{queue_names.size} queue(s): #{queue_names.join(", ")}" }
  queue_names.each { |name| ensure_queue(name) }
end

#ensure_dead_letter_queue(name) ⇒ Object 



322
323
324
325
326
327
328
329
330
331
332
333
334
 
# File 'lib/pgbus/client.rb', line 322

def ensure_dead_letter_queue(name)
  dlq_name = config.dead_letter_queue_name(name)
  return if @queues_created[dlq_name]

  if queue_ddl_rides_caller_transaction?
    create_dead_letter_queue_physically(dlq_name)
  else
    @queues_created.compute_if_absent(dlq_name) do
      create_dead_letter_queue_physically(dlq_name)
      true
    end
  end
end

#ensure_queue(name) ⇒ Object 



311
312
313
314
 
# File 'lib/pgbus/client.rb', line 311

def ensure_queue(name)
  ensure_pgmq_schema
  @queue_strategy.physical_queue_names(name).each { |pq| ensure_single_queue(pq) }
end

#in_recovery? ⇒ Boolean

Whether the job connection currently lands on a read-only replica (pg_is_in_recovery() => t). Used by the doctor to warn about a read/write-splitting pooler that could route pgmq's VOLATILE read/archive to a standby, silently stalling job processing (issue #332). Raw PG error propagates so the caller can render the reason.

Returns:

  • (Boolean) 


243
244
245
246
247
 
# File 'lib/pgbus/client.rb', line 243

def in_recovery?
  with_raw_connection do |conn|
    conn.exec(Process::PrimaryValidator::RECOVERY_QUERY).getvalue(0, 0) == "t"
  end
end

#list_notify_insert_throttles ⇒ Object 



791
792
793
794
795
 
# File 'lib/pgbus/client.rb', line 791

def list_notify_insert_throttles
  with_stale_connection_retry do
    synchronized { @pgmq.list_notify_insert_throttles }
  end
end

#list_queues ⇒ Object 



641
642
643
644
645
 
# File 'lib/pgbus/client.rb', line 641

def list_queues
  with_stale_connection_retry do
    synchronized { @pgmq.list_queues }
  end
end

#message_exists?(queue_name, msg_id: nil, uniqueness_key: nil) ⇒ Boolean

Check whether a message exists in the given queue.

Pass either msg_id for a fast primary-key lookup, or uniqueness_key to scan the queue for any message whose payload carries that key in the pgbus_uniqueness_key JSONB field. The latter is used by the dispatcher reaper to determine if a uniqueness lock with msg_id=0 (placeholder) still has a corresponding queue message.

queue_name may be either a logical name (e.g. "default") or an already prefixed physical name (e.g. "pgbus_default"). The client normalizes both.

Returns:

true  — the message definitely exists in the queue
false — the message definitely does not exist
nil   — could not determine (e.g. queue table missing or unknown error).
      Callers MUST treat nil as "exists" for safety.

Returns:

  • (Boolean) 


679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
 
# File 'lib/pgbus/client.rb', line 679

def message_exists?(queue_name, msg_id: nil, uniqueness_key: nil)
  has_msg_id = !msg_id.nil?
  has_uniqueness_key = !uniqueness_key.nil?
  raise ArgumentError, "pass exactly one of msg_id or uniqueness_key" unless has_msg_id ^ has_uniqueness_key

  full_name = resolve_full_queue_name(queue_name)
  sanitized = QueueNameValidator.sanitize!(full_name)

  synchronized do
    with_raw_connection do |conn|
      if has_msg_id
        msg_id_present?(conn, sanitized, msg_id.to_i)
      else
        uniqueness_key_present?(conn, sanitized, uniqueness_key)
      end
    end
  end
rescue ActiveRecord::StatementInvalid => e
  raise unless undefined_table_error?(e)

  nil
rescue StandardError => e
  raise unless defined?(PG::UndefinedTable) && e.is_a?(PG::UndefinedTable)

  nil
end

#metrics(queue_name = nil) ⇒ Object 



568
569
570
571
572
573
574
575
576
577
578
 
# File 'lib/pgbus/client.rb', line 568

def metrics(queue_name = nil)
  with_stale_connection_retry do
    synchronized do
      if queue_name
        @pgmq.metrics(config.queue_name(queue_name))
      else
        @pgmq.metrics_all
      end
    end
  end
end

#move_to_dead_letter(queue_name, message) ⇒ Object 



553
554
555
556
557
558
559
560
561
562
563
564
565
566
 
# File 'lib/pgbus/client.rb', line 553

def move_to_dead_letter(queue_name, message)
  dlq_name = config.dead_letter_queue_name(queue_name)
  full_queue = config.queue_name(queue_name)

  with_stale_connection_retry do
    ensure_dead_letter_queue(queue_name)
    synchronized do
      @pgmq.transaction do |txn|
        txn.produce(dlq_name, message.message, headers: message.headers)
        txn.delete(full_queue, message.msg_id.to_i)
      end
    end
  end
end

#notify_enabled?(queue_name) ⇒ Boolean

Whether the given logical queue currently has a live PGMQ insert-NOTIFY trigger with pgbus's throttle interval on every physical table it maps to. Uses the same physical-name resolution as bootstrap (@queue_strategy), so a priority queue's _p0.._pN sub-tables — where the trigger actually lives — are all checked, not the bare prefixed name that priority mode never creates. Returns false when any physical table lacks the trigger or the check can't run.

Returns:

  • (Boolean) 


264
265
266
267
 
# File 'lib/pgbus/client.rb', line 264

def notify_enabled?(queue_name)
  names = @queue_strategy.physical_queue_names(queue_name)
  names.all? { |physical| notify_trigger_current?(physical, NOTIFY_THROTTLE_MS) }
end

#oldest_claimable_ages(queue_name = nil) ⇒ Object

Age (seconds) of the oldest message actually eligible for pickup, i.e. whose visibility timeout has elapsed. Unlike pgmq's oldest_msg_age_sec (computed from enqueued_at), a scheduled or backoff-parked message — future vt — contributes nothing until it comes due, so a queue holding only parked messages reads nil ("no claimable backlog") instead of an age growing at wall-clock rate (issue #389). pgmq's metrics_result type is frozen upstream, so this lives here rather than in the SQL function.

With a queue name: the age for that (prefixed) queue, or nil. Without: a hash of every physical queue in pgmq.meta to its age.

Routes through the pooled @pgmq.with_connection (health-checked, bounded by the statement/socket timeouts applied at Client#initialize) rather than a fresh unbounded PG.connect per call — same rationale as notify_trigger_current?. synchronized: on the shared-Proc path @pgmq rides the AR raw connection, so the query must serialize against concurrent PGMQ operations. One checkout spans all per-queue queries; nothing nests inside it, so the shared pool_size=1 path is safe.



598
599
600
601
602
603
604
605
606
607
608
609
610
 
# File 'lib/pgbus/client.rb', line 598

def oldest_claimable_ages(queue_name = nil)
  synchronized do
    @pgmq.with_connection do |conn|
      if queue_name
        claimable_age_for(conn, config.queue_name(queue_name))
      else
        names = conn.exec("SELECT queue_name FROM pgmq.meta ORDER BY queue_name")
                    .map { |row| row["queue_name"] }
        names.to_h { |name| [name, claimable_age_for(conn, name)] }
      end
    end
  end
end

#pgmq_installed? ⇒ Boolean

Whether the PGMQ schema itself is present (the pgmq.meta table exists), independent of pgbus's own version-tracking table. Lets a caller tell "PGMQ installed via the extension / before version tracking" (schema present, no tracking row) apart from "PGMQ not installed at all".

Returns:

  • (Boolean) 


282
283
284
285
286
 
# File 'lib/pgbus/client.rb', line 282

def pgmq_installed?
  with_raw_connection do |conn|
    conn.exec(PGMQ_META_CHECK_SQL).ntuples.positive?
  end
end

#pgmq_schema_version ⇒ Object

The most recently recorded installed PGMQ schema version string (e.g. "1.5.0"), read from the pgbus_pgmq_schema_versions tracking table. Returns nil when nothing is tracked yet or the table does not exist — the same logic the pgbus:pgmq:status rake task uses, kept here so the doctor and the rake task share one raw-SQL path (never SQL outside the Client).



293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
 
# File 'lib/pgbus/client.rb', line 293

def pgmq_schema_version
  with_raw_connection do |conn|
    result = conn.exec(
      "SELECT version FROM pgbus_pgmq_schema_versions ORDER BY installed_at DESC LIMIT 1"
    )
    row = result.first
    row && row["version"]
  end
rescue ActiveRecord::StatementInvalid => e
  raise unless undefined_table_error?(e)

  nil
rescue StandardError => e
  raise unless defined?(PG::UndefinedTable) && e.is_a?(PG::UndefinedTable)

  nil
end

#physical_queue_names(logical_name) ⇒ Object

The physical PGMQ queue table names a logical queue maps to — one for a standard queue, or the _p0.._pN sub-queues when priority is enabled. This is the SAME resolution the bootstrap path uses (@queue_strategy), so a caller diffing configured-vs-existing queues compares the exact names PGMQ actually holds rather than the bare prefixed name.



274
275
276
 
# File 'lib/pgbus/client.rb', line 274

def physical_queue_names(logical_name)
  @queue_strategy.physical_queue_names(logical_name)
end

#ping ⇒ Object

Lightweight liveness probe used by the doctor: open a raw connection and run SELECT 1. Unlike verify_connection! (which wraps failures as ConfigurationError for the supervisor boot path), ping lets the raw PG/PGMQ error propagate so the caller can render the underlying reason. Returns true on success; a bad connection raises rather than returning false — the caller renders the underlying reason — so this is a probe, not a boolean predicate, hence no ? suffix.



233
234
235
236
 
# File 'lib/pgbus/client.rb', line 233

def ping # rubocop:disable Naming/PredicateMethod
  with_raw_connection { |conn| conn.exec("SELECT 1") }
  true
end

#pool_stats ⇒ Object

Snapshot of the PGMQ connection pool: available:, pool_timeout:.

Reads pgmq-ruby's own pool counters (@pgmq.stats -> available:) and adds the configured pool_timeout so alerting has the full picture: how many connections exist, how many are free right now, and how long a checkout waits before raising a pool-timeout error. Works on both the dedicated-pool path and the shared-Proc path (where size is 1).

Purely observational — wrapped in a rescue that returns {} so a probe or heartbeat reading the pool can never break job processing. Not routed through with_stale_connection_retry: reading in-memory counters touches no socket, and a failing read must degrade to {} rather than retry.



624
625
626
627
628
629
 
# File 'lib/pgbus/client.rb', line 624

def pool_stats
  @pgmq.stats.merge(pool_timeout: config.pool_timeout)
rescue StandardError => e
  Pgbus.logger.debug { "[Pgbus::Client] pool_stats unavailable: #{e.class}: #{e.message}" }
  {}
end

#publish_to_topic(routing_key, payload, headers: nil, delay: 0) ⇒ Object 



823
824
825
826
827
828
829
830
831
832
833
834
 
# File 'lib/pgbus/client.rb', line 823

def publish_to_topic(routing_key, payload, headers: nil, delay: 0)
  with_stale_connection_retry do
    synchronized do
      @pgmq.produce_topic(
        routing_key,
        serialize(payload),
        headers: headers && serialize(headers),
        delay: delay
      )
    end
  end
end

#purge_archive(queue_name, older_than:, batch_size: 1000) ⇒ Object 



706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
 
# File 'lib/pgbus/client.rb', line 706

def purge_archive(queue_name, older_than:, batch_size: 1000)
  full_name = config.queue_name(queue_name)
  sanitized = QueueNameValidator.sanitize!(full_name)
  total = 0

  sql = "DELETE FROM pgmq.a_#{sanitized} " \
        "WHERE ctid = ANY(ARRAY(SELECT ctid FROM pgmq.a_#{sanitized} WHERE enqueued_at < $1 LIMIT $2))"

  loop do
    deleted = synchronized do
      with_raw_connection do |conn|
        conn.exec_params(sql, [older_than, batch_size]).cmd_tuples
      end
    end
    total += deleted
    break if deleted < batch_size
  end

  total
end

#purge_queue(queue_name, prefixed: true) ⇒ Object 



647
648
649
650
651
652
 
# File 'lib/pgbus/client.rb', line 647

def purge_queue(queue_name, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.purge_queue(name) }
  end
end

#read_batch(queue_name, qty:, vt: nil) ⇒ Object 



400
401
402
403
404
405
406
407
408
409
 
# File 'lib/pgbus/client.rb', line 400

def read_batch(queue_name, qty:, vt: nil)
  full_name = config.queue_name(queue_name)
  guarded_read do
    Instrumentation.instrument("pgbus.client.read_batch", queue: full_name, qty: qty) do
      with_stale_connection_retry do
        synchronized { with_read_timeout { @pgmq.read_batch(full_name, vt: vt || config.visibility_timeout, qty: qty) } }
      end
    end
  end
end

#read_batch_prioritized(queue_name, qty:, vt: nil) ⇒ Object

Read from priority sub-queues, highest priority (p0) first. Returns [priority_queue_name, messages] pairs.



413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
 
# File 'lib/pgbus/client.rb', line 413

def read_batch_prioritized(queue_name, qty:, vt: nil)
  # Non-priority fast path delegates to read_batch, which is already gated
  # by the connection-health breaker — no extra guard needed here.
  unless @queue_strategy.priority?
    return (read_batch(queue_name, qty: qty, vt: vt) || []).map do |m|
      [config.queue_name(queue_name), m]
    end
  end

  # The priority loop issues its own reads, so gate the whole loop: an open
  # breaker fails fast before any sub-queue is touched, and the loop as a
  # unit records one success/failure with the latch.
  guarded_read do
    remaining = qty
    results = []

    config.priority_queue_names(queue_name).each do |pq_name|
      break if remaining <= 0

      msgs = Instrumentation.instrument("pgbus.client.read_batch", queue: pq_name, qty: remaining) do
        with_stale_connection_retry do
          synchronized { with_read_timeout { @pgmq.read_batch(pq_name, vt: vt || config.visibility_timeout, qty: remaining) } }
        end
      end || []

      msgs.each { |m| results << [pq_name, m] }
      remaining -= msgs.size
    end

    results
  end
end

#read_grouped(queue_name, qty:, vt: nil) ⇒ Object

--- Grouped reads (PGMQ v1.11.0+) ---



729
730
731
732
733
734
735
736
737
738
 
# File 'lib/pgbus/client.rb', line 729

def read_grouped(queue_name, qty:, vt: nil)
  full_name = config.queue_name(queue_name)
  guarded_read do
    Instrumentation.instrument("pgbus.client.read_grouped", queue: full_name, qty: qty) do
      with_stale_connection_retry do
        synchronized { with_read_timeout { @pgmq.read_grouped(full_name, vt: vt || config.visibility_timeout, qty: qty) } }
      end
    end
  end
end

#read_grouped_head(queue_name, qty:, vt: nil) ⇒ Object 



751
752
753
754
755
756
757
758
 
# File 'lib/pgbus/client.rb', line 751

def read_grouped_head(queue_name, qty:, vt: nil)
  full_name = config.queue_name(queue_name)
  guarded_read do
    with_stale_connection_retry do
      synchronized { with_read_timeout { @pgmq.read_grouped_head(full_name, vt: vt || config.visibility_timeout, qty: qty) } }
    end
  end
end

#read_grouped_rr(queue_name, qty:, vt: nil) ⇒ Object 



740
741
742
743
744
745
746
747
748
749
 
# File 'lib/pgbus/client.rb', line 740

def read_grouped_rr(queue_name, qty:, vt: nil)
  full_name = config.queue_name(queue_name)
  guarded_read do
    Instrumentation.instrument("pgbus.client.read_grouped_rr", queue: full_name, qty: qty) do
      with_stale_connection_retry do
        synchronized { with_read_timeout { @pgmq.read_grouped_rr(full_name, vt: vt || config.visibility_timeout, qty: qty) } }
      end
    end
  end
end

#read_message(queue_name, vt: nil) ⇒ Object 



389
390
391
392
393
394
395
396
397
398
 
# File 'lib/pgbus/client.rb', line 389

def read_message(queue_name, vt: nil)
  full_name = config.queue_name(queue_name)
  guarded_read do
    Instrumentation.instrument("pgbus.client.read_message", queue: full_name) do
      with_stale_connection_retry do
        synchronized { with_read_timeout { @pgmq.read(full_name, vt: vt || config.visibility_timeout) } }
      end
    end
  end
end

#read_multi(queue_names, qty:, vt: nil, limit: nil) ⇒ Object

Read from multiple queues in a single SQL query (UNION ALL). Each returned message includes a queue_name field identifying its source. queue_names should be logical names (prefix is added automatically).

qty is the per-queue cap (pgmq-ruby semantics), so without limit: the caller receives up to queue_count * qty messages. Pass limit: to cap the total across all queues — required when feeding a fixed-size pool, otherwise the pool can overflow on multi-queue reads (issue #123).

STRICT-PRIORITY CONTRACT (issue #381): when limit: is smaller than the total available, earlier-listed queues win — the capsule DSL's "list order = strict priority" promise rides on this. The mechanism is incidental: pgmq-ruby builds pgmq.read(q1) UNION ALL pgmq.read(q2) … LIMIT n, and Postgres's Append node fills the LIMIT from the subqueries in written order. Nothing upstream promises that, so the contract is pinned by spec/integration/multi_queue_priority_spec.rb — if that canary ever breaks, switch callers to ordered per-queue reads (the Worker#fetch_prioritized pattern) instead of relying on this method.

vt-claim caveat: each subquery may claim (set vt on) up to qty rows even when the outer LIMIT discards them — a discarded row goes invisible for one visibility timeout without being processed. Size qty/limit accordingly on latency-sensitive queues.



486
487
488
489
490
491
492
493
494
495
496
497
498
499
 
# File 'lib/pgbus/client.rb', line 486

def read_multi(queue_names, qty:, vt: nil, limit: nil)
  full_names = queue_names.map { |q| config.queue_name(q) }
  guarded_read do
    Instrumentation.instrument("pgbus.client.read_multi", queues: full_names, qty: qty, limit: limit) do
      with_stale_connection_retry do
        synchronized do
          with_read_timeout do
            @pgmq.read_multi(full_names, vt: vt || config.visibility_timeout, qty: qty, limit: limit)
          end
        end
      end
    end
  end
end

#read_with_poll(queue_name, qty:, vt: nil, max_poll_seconds: 5, poll_interval_ms: 100) ⇒ Object 



446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
 
# File 'lib/pgbus/client.rb', line 446

def read_with_poll(queue_name, qty:, vt: nil, max_poll_seconds: 5, poll_interval_ms: 100)
  full_name = config.queue_name(queue_name)
  guarded_read do
    with_stale_connection_retry do
      synchronized do
        @pgmq.read_with_poll(
          full_name,
          vt: vt || config.visibility_timeout,
          qty: qty,
          max_poll_seconds: max_poll_seconds,
          poll_interval_ms: poll_interval_ms
        )
      end
    end
  end
end

#reload ⇒ Object

Operator escape hatch (issue #354): drop every pooled PGMQ connection — job pool AND the live streams pool — and let the pools rebuild lazily on next checkout (pgmq-ruby >= 0.7.1). Use to recover connections libpq still reports as CONNECTION_OK but that are in fact wedged (e.g. after a wall-clock interrupt cut a query mid-flight), which pgmq-ruby's checkout health check cannot detect. Unlike #close, the pools stay usable. Connections checked out by other threads mid-reload are unaffected.

No-op (returns false) on the shared-AR Proc path: those pool slots wrap ActiveRecord's own raw connection — reloading would close AR's socket out from under the application. Returns true after a reload.



891
892
893
894
895
896
897
898
899
900
901
902
903
904
 
# File 'lib/pgbus/client.rb', line 891

def reload # rubocop:disable Naming/PredicateMethod -- command that reports whether it acted, like #ping
  if @shared_connection
    Pgbus.logger.warn do
      "[Pgbus::Client] reload skipped: pgbus is sharing ActiveRecord's connection " \
        "(Proc connection_options) and won't close a socket it doesn't own. " \
        "Manage that connection through ActiveRecord instead."
    end
    return false
  end

  @pgmq.reload
  @streams_pool.reload
  true
end

#resize_streams_pool(new_size) ⇒ ResizablePool::SwapStats

Opt-in hot-swap of the dedicated streams pool to a new size (issue #323 spike). Builds a fresh PGMQ::Client at new_size with the SAME bounds-applied connection options, atomically swaps the live reference, then drains + closes the old pool (bounded, never Thread#kill). NOT called automatically — there is no control loop here; a caller triggers it explicitly.

No-op on the shared-AR (Proc) path (the streams pool aliases the job pool, which is non-thread-safe and forced to pool_size 1 — swapping it would corrupt the job pool), and no-op when the size is unchanged.

Returns:

Raises:

  • (ArgumentError) 


862
863
864
865
866
867
868
869
870
871
872
 
# File 'lib/pgbus/client.rb', line 862

def resize_streams_pool(new_size)
  raise ArgumentError, "new_size must be a positive integer" unless new_size.is_a?(Integer) && new_size.positive?
  return { swapped: false, reason: :shared_connection } if @shared_connection
  return { swapped: false, reason: :unchanged } if streams_pool.stats[:size] == new_size

  from_size = streams_pool.stats[:size]
  new_pgmq = PGMQ::Client.new(
    @streams_conn_opts, pool_size: new_size, pool_timeout: config.streams_pool_timeout
  )
  @streams_pool.swap(new_pgmq, from_size: from_size, to_size: new_size)
end

#send_batch(queue_name, payloads, headers: nil, delay: 0) ⇒ Object 



378
379
380
381
382
383
384
385
386
387
 
# File 'lib/pgbus/client.rb', line 378

def send_batch(queue_name, payloads, headers: nil, delay: 0)
  full_name = config.queue_name(queue_name)
  serialized, serialized_headers = serialize_batch(payloads, headers)
  Instrumentation.instrument("pgbus.client.send_batch", queue: full_name, size: payloads.size) do
    with_stale_connection_retry do
      ensure_queue(queue_name)
      synchronized { @pgmq.produce_batch(full_name, serialized, headers: serialized_headers, delay: delay) }
    end
  end
end

#send_message(queue_name, payload, headers: nil, delay: 0, priority: nil) ⇒ Object 



336
337
338
339
340
341
342
343
344
 
# File 'lib/pgbus/client.rb', line 336

def send_message(queue_name, payload, headers: nil, delay: 0, priority: nil)
  target = @queue_strategy.target_queue(queue_name, priority)
  Instrumentation.instrument("pgbus.client.send_message", queue: target) do
    with_stale_connection_retry do
      ensure_queue(queue_name)
      synchronized { @pgmq.produce(target, serialize(payload), headers: headers && serialize(headers), delay: delay) }
    end
  end
end

#send_stream_message(stream_name, payload, headers: nil, delay: 0) ⇒ Object

Durable stream broadcast. Unlike #send_message, this ALWAYS targets the bare queue (config.queue_name) and never the priority strategy's _p0.._pN sub-queues: streams are delivered by a non-consuming peek (read_after) on the bare queue, and the streamer LISTENs on the bare channel, so a broadcast routed to _p1 would never reach the browser (issue #310). ensure_stream_queue creates the bare queue + NOTIFY trigger + archive index, mirroring this bare-name write path.



353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
 
# File 'lib/pgbus/client.rb', line 353

def send_stream_message(stream_name, payload, headers: nil, delay: 0)
  target = config.queue_name(stream_name)
  # Capture the produced msg_id — it is this method's return value (callers
  # like Stream#broadcast rely on it), so the autoscale trigger below must NOT
  # become the last expression.
  msg_id = Instrumentation.instrument("pgbus.client.send_message", queue: target) do
    with_stale_connection_retry do
      ensure_stream_queue(stream_name)
      # Publish through the dedicated streams pool (issue #315) so a
      # saturated job pool can't block a broadcast on pool checkout. On the
      # shared-AR path @streams_pgmq aliases @pgmq and synchronized still
      # serializes on the mutex.
      synchronized do
        streams_pool.produce(target, serialize(payload), headers: headers && serialize(headers), delay: delay)
      end
    end
  end
  # Opportunistically autoscale the streams pool from the publish path so a
  # pure-publisher process (no streamer) still grows under a broadcast storm
  # (issue #323 follow-up). Throttled + fail-soft — never delays or breaks the
  # broadcast; nil (a no-op) unless autoscale is on and the pool is dedicated.
  streams_pool_trigger&.maybe_check
  msg_id
end

#set_visibility_timeout(queue_name, msg_id, vt:, prefixed: true) ⇒ Object

Set visibility timeout. Pass prefixed: false when queue_name is already the full PGMQ queue name.



537
538
539
540
541
542
 
# File 'lib/pgbus/client.rb', line 537

def set_visibility_timeout(queue_name, msg_id, vt:, prefixed: true)
  name = prefixed ? config.queue_name(queue_name) : queue_name
  with_stale_connection_retry do
    synchronized { @pgmq.set_vt(name, msg_id, vt: vt) }
  end
end

#shared_connection? ⇒ Boolean

True when this client shares ActiveRecord's connection (the Proc connection_options path): pool_size is forced to 1 and every operation is serialized through @pgmq_mutex. False on the dedicated-connection path, where pgmq-ruby owns its own pool and no mutex is needed.

Returns:

  • (Boolean) 


182
183
184
 
# File 'lib/pgbus/client.rb', line 182

def shared_connection?
  @shared_connection
end

#streams_pool_stats ⇒ Object

Same shape as #pool_stats but for the dedicated streams pool (issue #315). On the shared-AR path @streams_pgmq aliases @pgmq, so this reports the job pool's counters — accurate, since streams share that connection there.



634
635
636
637
638
639
 
# File 'lib/pgbus/client.rb', line 634

def streams_pool_stats
  streams_pool.stats.merge(pool_timeout: config.streams_pool_timeout)
rescue StandardError => e
  Pgbus.logger.debug { "[Pgbus::Client] streams_pool_stats unavailable: #{e.class}: #{e.message}" }
  {}
end

#streams_swap_stats ⇒ Object

Accumulated streams-pool swap telemetry (issue #323) — for the bench and a future control loop. Zero-valued before any swap.



876
877
878
 
# File 'lib/pgbus/client.rb', line 876

def streams_swap_stats
  @streams_pool.stats_snapshot
end

#synchronizing? ⇒ Boolean

Whether the shared-connection serialization mutex is currently held. False on the dedicated-connection path (no mutex). Lets callers assert that a code path (e.g. a retry backoff sleep) runs OUTSIDE the mutex without reaching into the mutex object itself.

Returns:

  • (Boolean) 


190
191
192
 
# File 'lib/pgbus/client.rb', line 190

def synchronizing?
  @pgmq_mutex ? @pgmq_mutex.locked? : false
end

#transaction(&block) ⇒ Object

Open a PGMQ transaction. The caller block may run twice if the first attempt hits a pre-flight stale-connection error — safe because no SQL was sent on the first attempt (the connection was dead before the BEGIN).



547
548
549
550
551
 
# File 'lib/pgbus/client.rb', line 547

def transaction(&block)
  with_stale_connection_retry do
    synchronized { @pgmq.transaction(&block) }
  end
end

#update_notify_insert(queue_name, throttle_interval_ms:) ⇒ Object 



784
785
786
787
788
789
 
# File 'lib/pgbus/client.rb', line 784

def update_notify_insert(queue_name, throttle_interval_ms:)
  full_name = config.queue_name(queue_name)
  with_stale_connection_retry do
    synchronized { @pgmq.update_notify_insert(full_name, throttle_interval_ms: throttle_interval_ms) }
  end
end

#verify_connection! ⇒ Object

Actively open a database connection and run SELECT 1 so a bad database_url / connection_params surfaces at boot instead of on the first operation. PGMQ::Client's pool is lazy — nothing touches the database at init — so without this the supervisor forks children that crash-loop against an unreachable DB. Called from Supervisor#run before any queue bootstrap or forking.

Raises Pgbus::ConfigurationError (not a transient PGMQ error) because a failure here means the operator's connection config is wrong: the message carries the underlying error plus which config source was in use.



204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/pgbus/client.rb', line 204

def verify_connection!
  synchronized do
    @pgmq.with_connection do |conn|
      conn.exec("SELECT 1")
      # When require_primary is set, reject a connection that landed on a
      # read-only replica at boot rather than letting a read/write-splitting
      # pooler silently route pgmq's VOLATILE read/archive to a standby,
      # where workers read nothing and jobs stop with a healthy heartbeat
      # (issue #332). Off by default, so a single-primary deployment is
      # unaffected.
      Process::PrimaryValidator.validate_primary!(conn) if config.require_primary
    end
  end
  true
rescue Process::ReplicaConnectionError => e
  raise ConfigurationError,
        "Database connection via #{connection_source} landed on a read-only replica " \
        "(require_primary is set): #{e.message}"
rescue PGMQ::Errors::ConnectionError, PG::Error => e
  raise ConfigurationError, "Database connection failed via #{connection_source}: #{e.message}"
end

#wait_for_notify(queue_name, timeout: nil, &block) ⇒ Object

--- LISTEN/NOTIFY management (PGMQ v1.11.0+) ---



777
778
779
780
781
782
 
# File 'lib/pgbus/client.rb', line 777

def wait_for_notify(queue_name, timeout: nil, &block)
  full_name = config.queue_name(queue_name)
  with_stale_connection_retry do
    synchronized { @pgmq.wait_for_notify(full_name, timeout: timeout, &block) }
  end
end