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

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
 
# 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_connection_options (which defaults
    # to connection_options but honors streams_database_url/host/port for a
    # separate/direct streams DB — issue #315), 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_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.



469
470
471
472
473
474
 
# File 'lib/pgbus/client.rb', line 469

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.



461
462
463
464
465
466
 
# File 'lib/pgbus/client.rb', line 461

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



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

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 



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

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.



215
216
217
 
# File 'lib/pgbus/client.rb', line 215

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



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

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



679
680
681
682
683
684
 
# File 'lib/pgbus/client.rb', line 679

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 



686
687
688
689
690
 
# File 'lib/pgbus/client.rb', line 686

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.



477
478
479
480
481
482
 
# File 'lib/pgbus/client.rb', line 477

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



452
453
454
455
456
457
 
# File 'lib/pgbus/client.rb', line 452

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 



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

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 



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

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 



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

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 



276
277
278
279
 
# File 'lib/pgbus/client.rb', line 276

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) 


205
206
207
208
209
 
# File 'lib/pgbus/client.rb', line 205

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 



708
709
710
711
712
 
# File 'lib/pgbus/client.rb', line 708

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

#list_queues ⇒ Object 



558
559
560
561
562
 
# File 'lib/pgbus/client.rb', line 558

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) 


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

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 



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

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 



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

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) 


226
227
228
229
 
# File 'lib/pgbus/client.rb', line 226

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) 


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

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



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

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.



236
237
238
 
# File 'lib/pgbus/client.rb', line 236

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.



195
196
197
198
 
# File 'lib/pgbus/client.rb', line 195

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.



541
542
543
544
545
546
 
# File 'lib/pgbus/client.rb', line 541

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 



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

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 



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

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 



564
565
566
567
568
569
 
# File 'lib/pgbus/client.rb', line 564

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 



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

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.



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

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



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

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 



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

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 



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

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 



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

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



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

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 



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

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

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


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

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 



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

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 



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

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.



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

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.



486
487
488
489
490
491
 
# File 'lib/pgbus/client.rb', line 486

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) 


144
145
146
 
# File 'lib/pgbus/client.rb', line 144

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.



551
552
553
554
555
556
 
# File 'lib/pgbus/client.rb', line 551

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.



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

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) 


152
153
154
 
# File 'lib/pgbus/client.rb', line 152

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



496
497
498
499
500
 
# File 'lib/pgbus/client.rb', line 496

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

#update_notify_insert(queue_name, throttle_interval_ms:) ⇒ Object 



701
702
703
704
705
706
 
# File 'lib/pgbus/client.rb', line 701

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.



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

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



694
695
696
697
698
699
 
# File 'lib/pgbus/client.rb', line 694

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