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

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 ReadAfter

ReadAfter::DEFAULT_LIMIT

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.



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
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
 
# File 'lib/pgbus/client.rb', line 49

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

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.



39
40
41
42
43
44
 
# File 'lib/pgbus/client.rb', line 39

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.



471
472
473
474
475
476
 
# File 'lib/pgbus/client.rb', line 471

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.



463
464
465
466
467
468
 
# File 'lib/pgbus/client.rb', line 463

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



734
735
736
737
738
739
740
 
# File 'lib/pgbus/client.rb', line 734

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 



755
756
757
758
759
760
761
762
763
764
765
766
767
768
 
# File 'lib/pgbus/client.rb', line 755

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.



217
218
219
 
# File 'lib/pgbus/client.rb', line 217

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) ---



718
719
720
721
722
723
724
725
726
727
728
729
730
731
 
# File 'lib/pgbus/client.rb', line 718

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+) ---



681
682
683
684
685
686
 
# File 'lib/pgbus/client.rb', line 681

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 



688
689
690
691
692
 
# File 'lib/pgbus/client.rb', line 688

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.



479
480
481
482
483
484
 
# File 'lib/pgbus/client.rb', line 479

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).



454
455
456
457
458
459
 
# File 'lib/pgbus/client.rb', line 454

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 



573
574
575
576
577
578
579
580
 
# File 'lib/pgbus/client.rb', line 573

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 



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

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 



289
290
291
292
293
294
295
296
297
298
299
300
 
# File 'lib/pgbus/client.rb', line 289

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

  @queues_created.compute_if_absent(dlq_name) do
    synchronized do
      @pgmq.create(dlq_name)
      tune_autovacuum(dlq_name)
    end
    true
  end
end

#ensure_queue(name) ⇒ Object 



278
279
280
281
 
# File 'lib/pgbus/client.rb', line 278

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) 


207
208
209
210
211
 
# File 'lib/pgbus/client.rb', line 207

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 



710
711
712
713
714
 
# File 'lib/pgbus/client.rb', line 710

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

#list_queues ⇒ Object 



560
561
562
563
564
 
# File 'lib/pgbus/client.rb', line 560

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) 


598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
 
# File 'lib/pgbus/client.rb', line 598

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 



519
520
521
522
523
524
525
526
527
528
529
 
# File 'lib/pgbus/client.rb', line 519

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 



504
505
506
507
508
509
510
511
512
513
514
515
516
517
 
# File 'lib/pgbus/client.rb', line 504

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) 


228
229
230
231
 
# File 'lib/pgbus/client.rb', line 228

def notify_enabled?(queue_name)
  names = @queue_strategy.physical_queue_names(queue_name)
  names.all? { |physical| notify_trigger_current?(physical, NOTIFY_THROTTLE_MS) }
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) 


246
247
248
249
250
251
252
253
 
# File 'lib/pgbus/client.rb', line 246

def pgmq_installed?
  with_raw_connection do |conn|
    result = conn.exec(
      "SELECT 1 FROM pg_tables WHERE schemaname = 'pgmq' AND tablename = 'meta' LIMIT 1"
    )
    result.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).



260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
 
# File 'lib/pgbus/client.rb', line 260

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.



238
239
240
 
# File 'lib/pgbus/client.rb', line 238

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.



197
198
199
200
 
# File 'lib/pgbus/client.rb', line 197

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.



543
544
545
546
547
548
 
# File 'lib/pgbus/client.rb', line 543

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 



742
743
744
745
746
747
748
749
750
751
752
753
 
# File 'lib/pgbus/client.rb', line 742

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 



625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
 
# File 'lib/pgbus/client.rb', line 625

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 



566
567
568
569
570
571
 
# File 'lib/pgbus/client.rb', line 566

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 



366
367
368
369
370
371
372
373
374
375
 
# File 'lib/pgbus/client.rb', line 366

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.



379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
 
# File 'lib/pgbus/client.rb', line 379

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+) ---



648
649
650
651
652
653
654
655
656
657
 
# File 'lib/pgbus/client.rb', line 648

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 



670
671
672
673
674
675
676
677
 
# File 'lib/pgbus/client.rb', line 670

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 



659
660
661
662
663
664
665
666
667
668
 
# File 'lib/pgbus/client.rb', line 659

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 



355
356
357
358
359
360
361
362
363
364
 
# File 'lib/pgbus/client.rb', line 355

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).



437
438
439
440
441
442
443
444
445
446
447
448
449
450
 
# File 'lib/pgbus/client.rb', line 437

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 



412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
 
# File 'lib/pgbus/client.rb', line 412

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.



810
811
812
813
814
815
816
817
818
819
820
821
822
823
 
# File 'lib/pgbus/client.rb', line 810

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) 


781
782
783
784
785
786
787
788
789
790
791
 
# File 'lib/pgbus/client.rb', line 781

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 



344
345
346
347
348
349
350
351
352
353
 
# File 'lib/pgbus/client.rb', line 344

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 



302
303
304
305
306
307
308
309
310
 
# File 'lib/pgbus/client.rb', line 302

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.



319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
 
# File 'lib/pgbus/client.rb', line 319

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.



488
489
490
491
492
493
 
# File 'lib/pgbus/client.rb', line 488

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) 


146
147
148
 
# File 'lib/pgbus/client.rb', line 146

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.



553
554
555
556
557
558
 
# File 'lib/pgbus/client.rb', line 553

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.



795
796
797
 
# File 'lib/pgbus/client.rb', line 795

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) 


154
155
156
 
# File 'lib/pgbus/client.rb', line 154

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).



498
499
500
501
502
 
# File 'lib/pgbus/client.rb', line 498

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

#update_notify_insert(queue_name, throttle_interval_ms:) ⇒ Object 



703
704
705
706
707
708
 
# File 'lib/pgbus/client.rb', line 703

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.



168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/pgbus/client.rb', line 168

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+) ---



696
697
698
699
700
701
 
# File 'lib/pgbus/client.rb', line 696

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