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
 
# 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 = 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 = 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.



443
444
445
446
447
448
 
# File 'lib/pgbus/client.rb', line 443

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.



435
436
437
438
439
440
 
# File 'lib/pgbus/client.rb', line 435

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



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

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 



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

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.



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

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



690
691
692
693
694
695
696
697
698
699
700
701
702
703
 
# File 'lib/pgbus/client.rb', line 690

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



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

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 



660
661
662
663
664
 
# File 'lib/pgbus/client.rb', line 660

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.



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

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



426
427
428
429
430
431
 
# File 'lib/pgbus/client.rb', line 426

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 



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

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 



255
256
257
258
259
 
# File 'lib/pgbus/client.rb', line 255

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 



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

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 



250
251
252
253
 
# File 'lib/pgbus/client.rb', line 250

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

#list_notify_insert_throttles ⇒ Object 



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

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

#list_queues ⇒ Object 



532
533
534
535
536
 
# File 'lib/pgbus/client.rb', line 532

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) 


570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
 
# File 'lib/pgbus/client.rb', line 570

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 



491
492
493
494
495
496
497
498
499
500
501
 
# File 'lib/pgbus/client.rb', line 491

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 



476
477
478
479
480
481
482
483
484
485
486
487
488
489
 
# File 'lib/pgbus/client.rb', line 476

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) 


200
201
202
203
 
# File 'lib/pgbus/client.rb', line 200

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) 


218
219
220
221
222
223
224
225
 
# File 'lib/pgbus/client.rb', line 218

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



232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
 
# File 'lib/pgbus/client.rb', line 232

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.



210
211
212
 
# File 'lib/pgbus/client.rb', line 210

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.



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

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.



515
516
517
518
519
520
 
# File 'lib/pgbus/client.rb', line 515

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 



714
715
716
717
718
719
720
721
722
723
724
725
 
# File 'lib/pgbus/client.rb', line 714

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 



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

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 



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

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 



338
339
340
341
342
343
344
345
346
347
 
# File 'lib/pgbus/client.rb', line 338

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.



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

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



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

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 



642
643
644
645
646
647
648
649
 
# File 'lib/pgbus/client.rb', line 642

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 



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

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 



327
328
329
330
331
332
333
334
335
336
 
# File 'lib/pgbus/client.rb', line 327

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



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

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 



384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
 
# File 'lib/pgbus/client.rb', line 384

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) 


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

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 



316
317
318
319
320
321
322
323
324
325
 
# File 'lib/pgbus/client.rb', line 316

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 



274
275
276
277
278
279
280
281
282
 
# File 'lib/pgbus/client.rb', line 274

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.



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

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.



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

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) 


142
143
144
 
# File 'lib/pgbus/client.rb', line 142

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.



525
526
527
528
529
530
 
# File 'lib/pgbus/client.rb', line 525

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.



767
768
769
 
# File 'lib/pgbus/client.rb', line 767

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) 


150
151
152
 
# File 'lib/pgbus/client.rb', line 150

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



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

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

#update_notify_insert(queue_name, throttle_interval_ms:) ⇒ Object 



675
676
677
678
679
680
 
# File 'lib/pgbus/client.rb', line 675

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.



164
165
166
167
168
169
170
171
 
# File 'lib/pgbus/client.rb', line 164

def verify_connection!
  synchronized do
    @pgmq.with_connection { |conn| conn.exec("SELECT 1") }
  end
  true
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+) ---



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

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